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`).

apmpackage/apm/0.3.0/manifest.yml

@@ -120,12 +120,6 @@ policy_templates:
             required: false
             show_user: false
             default: 1000
-          - name: sourcemap_api_key
-            type: text
-            title: RUM - API Key for Sourcemaps
-            required: false
-            description: API Key for sourcemap feature. Enter as <Id>:<API Key>
-            show_user: false
           - name: api_key_limit
             type: integer
             title: Maximum number of API Keys for Agent authentication

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,26 @@ 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)
+	if len(cfg.SourceMapConfigs) == 0 {
+		c, err := elasticsearch.NewClient(cfg.ESConfig)
+		if err != nil {
+			return nil, err
+		}
+		index := strings.ReplaceAll(cfg.IndexPattern, "%{[observer.version]}", beatInfo.Version)
+		return sourcemap.NewElasticsearchStore(c, index, 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)
+	return sourcemap.NewFleetStore(&c, cfg.ESConfig.APIKey, cfg.SourceMapConfigs, cfg.Cache.Expiration)
 }
 
 // WrapRunServerWithProcessors wraps runServer such that it wraps args.Reporter

beater/config/agentconfig.go

@@ -75,12 +75,12 @@ type Service struct {
 
 // 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}, " ")
+	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",
@@ -292,6 +293,14 @@ func TestUnpackConfig(t *testing.T) {
 				"rum": map[string]interface{}{
 					"enabled": true,
 					"source_mapping": map[string]interface{}{
+						"source_maps": []map[string]string{
+							{
+								"service.name":    "opbeans-rum",
+								"service.version": "1.2.3",
+								"bundle.filepath": "/test/e2e/general-usecase/bundle.js.map",
+								"sourcemap.url":   "http://somewhere.com/bundle.js.map",
+							},
+						},
 						"cache": map[string]interface{}{
 							"expiration": 7,
 						},
@@ -365,6 +374,14 @@ func TestUnpackConfig(t *testing.T) {
 						},
 						IndexPattern: "apm-*-sourcemap*",
 						ESConfig:     elasticsearch.DefaultConfig(),
+						SourceMapConfigs: []SourceMapConfig{
+							{
+								ServiceName:    "opbeans-rum",
+								ServiceVersion: "1.2.3",
+								BundleFilepath: "/test/e2e/general-usecase/bundle.js.map",
+								SourceMapURL:   "http://somewhere.com/bundle.js.map",
+							},
+						},
 					},
 					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,14 +80,13 @@ func (c *RumConfig) setup(log *logp.Logger, dataStreamsEnabled bool, outputESCfg
 		return errors.Wrapf(err, "Invalid regex for `exclude_from_grouping`: ")
 	}
 
-	var apiKey string
-	if c.SourceMapping.esConfigured {
-		if dataStreamsEnabled {
-			// when running under Fleet, the only setting configured is the api key
-			apiKey = c.SourceMapping.ESConfig.APIKey
-		} else {
-			return nil
-		}
+	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
 	}
 
 	// fall back to elasticsearch output configuration for sourcemap storage if possible
@@ -98,9 +98,6 @@ func (c *RumConfig) setup(log *logp.Logger, dataStreamsEnabled bool, outputESCfg
 	if err := outputESCfg.Unpack(c.SourceMapping.ESConfig); err != nil {
 		return errors.Wrap(err, "unpacking Elasticsearch config into Sourcemap config")
 	}
-	if c.SourceMapping.ESConfig.APIKey == "" {
-		c.SourceMapping.ESConfig.APIKey = apiKey
-	}
 	return nil
 }
 
@@ -115,10 +112,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,26 @@
+// 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
+
+// SourceMapConfig holds source map configuration information.
+type SourceMapConfig struct {
+	ServiceName    string `config:"service.name"`
+	ServiceVersion string `config:"service.version"`
+	BundleFilepath string `config:"bundle.filepath"`
+	SourceMapURL   string `config:"sourcemap.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

@@ -805,7 +805,7 @@ 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)
+	store, err := sourcemap.NewElasticsearchStore(test.ESClientWithValidSourcemap(t), "apm-*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,7 @@ 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)
+		store, err := sourcemap.NewElasticsearchStore(client, "foo", time.Minute)
 		require.NoError(t, err)
 
 		// transform with sourcemap store

model/stacktrace_frame_test.go

@@ -422,7 +422,7 @@ func TestLibraryFrame(t *testing.T) {
 }
 
 func testSourcemapStore(t *testing.T, client elasticsearch.Client) *sourcemap.Store {
-	store, err := sourcemap.NewStore(client, "apm-*sourcemap*", time.Minute)
+	store, err := sourcemap.NewElasticsearchStore(client, "apm-*sourcemap*", time.Minute)
 	require.NoError(t, err)
 	return store
 }

sourcemap/es_store.go

@@ -25,12 +25,14 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"time"
 
 	"github.com/pkg/errors"
 
 	"github.com/elastic/beats/v7/libbeat/logp"
 
 	"github.com/elastic/apm-server/elasticsearch"
+	logs "github.com/elastic/apm-server/log"
 	"github.com/elastic/apm-server/utility"
 )
 
@@ -65,6 +67,19 @@ type esSourcemapResponse struct {
 	} `json:"hits"`
 }
 
+// NewElasticsearchStore returns an instance of Store for interacting with
+// sourcemaps stored in ElasticSearch.
+func NewElasticsearchStore(
+	c elasticsearch.Client,
+	index string,
+	expiration time.Duration,
+) (*Store, error) {
+	logger := logp.NewLogger(logs.Sourcemap)
+	s := &esStore{c, index, logger}
+
+	return newStore(s, logger, expiration)
+}
+
 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 {