feat: fetch sourcemaps from Elasticsearch

internal/beater/beater.go

@@ -23,7 +23,6 @@ import (
 	"fmt"
 	"net"
 	"net/http"
-	"net/url"
 	"os"
 	"runtime"
 	"strings"
@@ -33,14 +32,12 @@ import (
 	"github.com/hashicorp/go-multierror"
 	"github.com/pkg/errors"
 	"go.elastic.co/apm/module/apmgrpc/v2"
-	"go.elastic.co/apm/module/apmhttp/v2"
 	"go.elastic.co/apm/v2"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
-	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/esleg/eslegclient"
 	"github.com/elastic/beats/v7/libbeat/instrumentation"
 	"github.com/elastic/beats/v7/libbeat/licenser"
@@ -51,8 +48,6 @@ import (
 	agentconfig "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/monitoring"
-	"github.com/elastic/elastic-agent-libs/transport"
-	"github.com/elastic/elastic-agent-libs/transport/tlscommon"
 	"github.com/elastic/go-docappender"
 	"github.com/elastic/go-ucfg"
 
@@ -330,20 +325,15 @@ func (s *Runner) Run(ctx context.Context) error {
 
 	var sourcemapFetcher sourcemap.Fetcher
 	if s.config.RumConfig.Enabled && s.config.RumConfig.SourceMapping.Enabled {
-		fetcher, err := newSourcemapFetcher(
+		fetcher, cleanup, err := newSourcemapFetcher(
 			s.config.RumConfig.SourceMapping, s.fleetConfig,
 			kibanaClient, newElasticsearchClient,
 		)
 		if err != nil {
 			return err
 		}
-		cachingFetcher, err := sourcemap.NewCachingFetcher(
-			fetcher, s.config.RumConfig.SourceMapping.Cache.Expiration,
-		)
-		if err != nil {
-			return err
-		}
-		sourcemapFetcher = cachingFetcher
+		defer cleanup()
+		sourcemapFetcher = fetcher
 	}
 
 	// Create the runServer function. We start with newBaseRunServer, and then
@@ -823,73 +813,37 @@ func newSourcemapFetcher(
 	fleetCfg *config.Fleet,
 	kibanaClient *kibana.Client,
 	newElasticsearchClient func(*elasticsearch.Config) (*elasticsearch.Client, error),
-) (sourcemap.Fetcher, error) {
-	// When running under Fleet we only fetch via Fleet Server.
-	if fleetCfg != nil {
-		var tlsConfig *tlscommon.TLSConfig
-		var err error
-		if fleetCfg.TLS.IsEnabled() {
-			if tlsConfig, err = tlscommon.LoadTLSConfig(fleetCfg.TLS); err != nil {
-				return nil, err
-			}
-		}
-
-		timeout := 30 * time.Second
-		dialer := transport.NetDialer(timeout)
-		tlsDialer := transport.TLSDialer(dialer, tlsConfig, timeout)
+) (sourcemap.Fetcher, context.CancelFunc, error) {
+	esClient, err := newElasticsearchClient(cfg.ESConfig)
+	if err != nil {
+		return nil, nil, err
+	}
 
-		client := *http.DefaultClient
-		client.Transport = apmhttp.WrapRoundTripper(&http.Transport{
-			Proxy:           http.ProxyFromEnvironment,
-			Dial:            dialer.Dial,
-			DialTLS:         tlsDialer.Dial,
-			TLSClientConfig: tlsConfig.ToConfig(),
-		})
+	// For standalone, we query both Kibana and Elasticsearch for backwards compatibility.
+	var chained sourcemap.ChainedFetcher
 
-		fleetServerURLs := make([]*url.URL, len(fleetCfg.Hosts))
-		for i, host := range fleetCfg.Hosts {
-			urlString, err := common.MakeURL(fleetCfg.Protocol, "", host, 8220)
-			if err != nil {
-				return nil, err
-			}
-			u, err := url.Parse(urlString)
-			if err != nil {
-				return nil, err
-			}
-			fleetServerURLs[i] = u
-		}
+	size := 128
+	invalidationChan := make(chan sourcemap.Identifier, size)
 
-		artifactRefs := make([]sourcemap.FleetArtifactReference, len(cfg.Metadata))
-		for i, meta := range cfg.Metadata {
-			artifactRefs[i] = sourcemap.FleetArtifactReference{
-				ServiceName:        meta.ServiceName,
-				ServiceVersion:     meta.ServiceVersion,
-				BundleFilepath:     meta.BundleFilepath,
-				FleetServerURLPath: meta.SourceMapURL,
-			}
-		}
+	index := strings.ReplaceAll(cfg.IndexPattern, "%{[observer.version]}", version.Version)
 
-		return sourcemap.NewFleetFetcher(
-			&client,
-			fleetCfg.AccessAPIKey,
-			fleetServerURLs,
-			artifactRefs,
-		)
+	esFetcher := sourcemap.NewElasticsearchFetcher(esClient, index)
+	cachingFetcher, err := sourcemap.NewCachingFetcher(esFetcher, invalidationChan, size)
+	if err != nil {
+		return nil, nil, err
 	}
+	metadataCachingFetcher := sourcemap.NewMetadataCachingFetcher(esClient, cachingFetcher, index, invalidationChan)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	metadataCachingFetcher.StartBackgroundSync(ctx)
+
+	chained = append(chained, metadataCachingFetcher)
 
-	// For standalone, we query both Kibana and Elasticsearch for backwards compatibility.
-	var chained sourcemap.ChainedFetcher
 	if kibanaClient != nil {
 		chained = append(chained, sourcemap.NewKibanaFetcher(kibanaClient))
 	}
-	esClient, err := newElasticsearchClient(cfg.ESConfig)
-	if err != nil {
-		return nil, err
-	}
-	index := strings.ReplaceAll(cfg.IndexPattern, "%{[observer.version]}", version.Version)
-	esFetcher := sourcemap.NewElasticsearchFetcher(esClient, index)
-	chained = append(chained, esFetcher)
-	return chained, nil
+
+	return chained, cancel, nil
 }
 
 // TODO: This is copying behavior from libbeat:

internal/beater/beater_test.go

@@ -18,7 +18,6 @@
 package beater
 
 import (
-	"compress/zlib"
 	"context"
 	"fmt"
 	"net/http"
@@ -40,6 +39,13 @@ func TestSourcemapIndexPattern(t *testing.T) {
 	test := func(t *testing.T, indexPattern, expected string) {
 		var requestPaths []string
 		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			// Ignore duplicates: we might have two requests since the first one
+			// will be from the init goroutine
+			for _, v := range requestPaths {
+				if v == r.URL.Path {
+					return
+				}
+			}
 			requestPaths = append(requestPaths, r.URL.Path)
 		}))
 		defer srv.Close()
@@ -51,11 +57,12 @@ func TestSourcemapIndexPattern(t *testing.T) {
 			cfg.RumConfig.SourceMapping.IndexPattern = indexPattern
 		}
 
-		fetcher, err := newSourcemapFetcher(
+		fetcher, cleanup, err := newSourcemapFetcher(
 			cfg.RumConfig.SourceMapping, nil,
 			nil, elasticsearch.NewClient,
 		)
 		require.NoError(t, err)
+		defer cleanup()
 		fetcher.Fetch(context.Background(), "name", "version", "path")
 		require.Len(t, requestPaths, 1)
 
@@ -65,7 +72,7 @@ func TestSourcemapIndexPattern(t *testing.T) {
 		assert.Equal(t, expected, path)
 	}
 	t.Run("default-pattern", func(t *testing.T) {
-		test(t, "", "apm-*-sourcemap*")
+		test(t, "", ".apm-source-map")
 	})
 	t.Run("with-observer-version", func(t *testing.T) {
 		test(t, "blah-%{[observer.version]}-blah", fmt.Sprintf("blah-%s-blah", version.Version))
@@ -89,11 +96,12 @@ func TestStoreUsesRUMElasticsearchConfig(t *testing.T) {
 	cfg.RumConfig.SourceMapping.ESConfig = elasticsearch.DefaultConfig()
 	cfg.RumConfig.SourceMapping.ESConfig.Hosts = []string{ts.URL}
 
-	fetcher, err := newSourcemapFetcher(
+	fetcher, cleanup, err := newSourcemapFetcher(
 		cfg.RumConfig.SourceMapping, nil,
 		nil, elasticsearch.NewClient,
 	)
 	require.NoError(t, err)
+	defer cleanup()
 	// Check that the provided rum elasticsearch config was used and
 	// Fetch() goes to the test server.
 	_, err = fetcher.Fetch(context.Background(), "app", "1.0", "/bundle/path")
@@ -102,41 +110,6 @@ func TestStoreUsesRUMElasticsearchConfig(t *testing.T) {
 	assert.True(t, called)
 }
 
-func TestFleetStoreUsed(t *testing.T) {
-	var called bool
-	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		called = true
-		wr := zlib.NewWriter(w)
-		defer wr.Close()
-		wr.Write([]byte(fmt.Sprintf(`{"sourceMap":%s}`, validSourcemap)))
-	}))
-	defer ts.Close()
-
-	cfg := config.DefaultConfig()
-	cfg.RumConfig.Enabled = true
-	cfg.RumConfig.SourceMapping.Enabled = true
-	cfg.RumConfig.SourceMapping.Metadata = []config.SourceMapMetadata{{
-		ServiceName:    "app",
-		ServiceVersion: "1.0",
-		BundleFilepath: "/bundle/path",
-		SourceMapURL:   "/my/path",
-	}}
-
-	fleetCfg := &config.Fleet{
-		Hosts:        []string{ts.URL[7:]},
-		Protocol:     "http",
-		AccessAPIKey: "my-key",
-		TLS:          nil,
-	}
-
-	fetcher, err := newSourcemapFetcher(cfg.RumConfig.SourceMapping, fleetCfg, nil, nil)
-	require.NoError(t, err)
-	_, err = fetcher.Fetch(context.Background(), "app", "1.0", "/bundle/path")
-	require.NoError(t, err)
-
-	assert.True(t, called)
-}
-
 func TestQueryClusterUUIDRegistriesExist(t *testing.T) {
 	stateRegistry := monitoring.GetNamespace("state").GetRegistry()
 	stateRegistry.Clear()

internal/beater/config/config_test.go

@@ -453,7 +453,7 @@ func TestUnpackConfig(t *testing.T) {
 						Cache: Cache{
 							Expiration: 7 * time.Second,
 						},
-						IndexPattern: "apm-*-sourcemap*",
+						IndexPattern: ".apm-source-map",
 						ESConfig:     elasticsearch.DefaultConfig(),
 						Metadata: []SourceMapMetadata{
 							{

internal/beater/config/rum.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/go-ucfg"
 
 	"github.com/elastic/apm-server/internal/elasticsearch"
 )
@@ -34,7 +35,7 @@ const (
 	defaultExcludeFromGrouping      = "^/webpack"
 	defaultLibraryPattern           = "node_modules|bower_components|~"
 	defaultSourcemapCacheExpiration = 5 * time.Minute
-	defaultSourcemapIndexPattern    = "apm-*-sourcemap*"
+	defaultSourcemapIndexPattern    = ".apm-source-map"
 	defaultSourcemapTimeout         = 5 * time.Second
 )
 
@@ -58,6 +59,7 @@ type SourceMapping struct {
 	Metadata     []SourceMapMetadata   `config:"metadata"`
 	Timeout      time.Duration         `config:"timeout" validate:"positive"`
 	esConfigured bool
+	es           *config.C
 }
 
 func (c *RumConfig) setup(log *logp.Logger, outputESCfg *config.C) error {
@@ -72,20 +74,35 @@ func (c *RumConfig) setup(log *logp.Logger, outputESCfg *config.C) error {
 		return errors.Wrapf(err, "Invalid regex for `exclude_from_grouping`: ")
 	}
 
-	// No need to unpack the ESConfig if SourceMapMetadata exist
 	if len(c.SourceMapping.Metadata) > 0 {
-		return nil
+		// We don't have the fleet fetcher anymore.
+		// Ignore metadata and setup the elasticsearch config.
+		log.Warn("Ignoring sourcemap metadata")
 	}
 
-	// fall back to elasticsearch output configuration for sourcemap storage if possible
 	if outputESCfg == nil {
 		log.Info("Unable to determine sourcemap storage, sourcemaps will not be applied")
 		return nil
 	}
-	log.Info("Falling back to elasticsearch output for sourcemap storage")
+
+	// Unpack the output elasticsearch config first
 	if err := outputESCfg.Unpack(c.SourceMapping.ESConfig); err != nil {
-		return errors.Wrap(err, "unpacking Elasticsearch config into Sourcemap config")
+		return errors.Wrap(err, "unpacking Elasticsearch output config into Sourcemap config")
+	}
+
+	// SourceMapping ES config not configured, use the main one and return early
+	if c.SourceMapping.es == nil {
+		log.Info("Using default sourcemap Elasticsearch config")
+		return nil
 	}
+
+	// Unpack the SourceMapping ES config on top of the output elasticsearch config
+	if err := c.SourceMapping.es.Unpack(c.SourceMapping.ESConfig); err != nil {
+		return errors.Wrap(err, "unpacking Elasticsearch sourcemap config into Sourcemap config")
+	}
+
+	c.SourceMapping.es = nil
+
 	return nil
 }
 
@@ -95,6 +112,11 @@ func (s *SourceMapping) Unpack(inp *config.C) error {
 		return errors.Wrap(err, "error unpacking sourcemapping config")
 	}
 	s.esConfigured = inp.HasField("elasticsearch")
+	var err error
+	var e ucfg.Error
+	if s.es, err = inp.Child("elasticsearch", -1); err != nil && (!errors.As(err, &e) || e.Reason() != ucfg.ErrMissing) {
+		return errors.Wrap(err, "error storing sourcemap elasticsearch config")
+	}
 	return nil
 }