feat: fetch sourcemaps from Elasticsearch

internal/beater/beater.go

@@ -23,24 +23,20 @@ import (
 	"fmt"
 	"net"
 	"net/http"
-	"net/url"
 	"os"
 	"runtime"
-	"strings"
 	"time"
 
 	"github.com/dustin/go-humanize"
 	"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 +47,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 +324,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(
-			s.config.RumConfig.SourceMapping, s.fleetConfig,
+		fetcher, cancel, err := newSourcemapFetcher(
+			s.config.RumConfig.SourceMapping,
 			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 cancel()
+		sourcemapFetcher = fetcher
 	}
 
 	// Create the runServer function. We start with newBaseRunServer, and then
@@ -818,78 +807,40 @@ func (s *Runner) newLibbeatFinalBatchProcessor(
 	return publisher, stop, nil
 }
 
+const sourcemapIndex = ".apm-source-map"
+
 func newSourcemapFetcher(
 	cfg config.SourceMapping,
-	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)
-
-		client := *http.DefaultClient
-		client.Transport = apmhttp.WrapRoundTripper(&http.Transport{
-			Proxy:           http.ProxyFromEnvironment,
-			Dial:            dialer.Dial,
-			DialTLS:         tlsDialer.Dial,
-			TLSClientConfig: tlsConfig.ToConfig(),
-		})
+) (sourcemap.Fetcher, context.CancelFunc, error) {
+	esClient, err := newElasticsearchClient(cfg.ESConfig)
+	if err != nil {
+		return nil, nil, err
+	}
 
-		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
-		}
+	var chained sourcemap.ChainedFetcher
 
-		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,
-			}
-		}
+	// start background sync job
+	ctx, cancel := context.WithCancel(context.Background())
+	metadataFetcher, invalidationChan := sourcemap.NewMetadataFetcher(ctx, esClient, sourcemapIndex)
 
-		return sourcemap.NewFleetFetcher(
-			&client,
-			fleetCfg.AccessAPIKey,
-			fleetServerURLs,
-			artifactRefs,
-		)
+	esFetcher := sourcemap.NewElasticsearchFetcher(esClient, sourcemapIndex)
+	size := 128
+	cachingFetcher, err := sourcemap.NewBodyCachingFetcher(esFetcher, size, invalidationChan)
+	if err != nil {
+		cancel()
+		return nil, nil, err
 	}
+	sourcemapFetcher := sourcemap.NewSourcemapFetcher(metadataFetcher, cachingFetcher)
+
+	chained = append(chained, sourcemapFetcher)
 
-	// 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,123 +18,125 @@
 package beater
 
 import (
+	"bytes"
 	"compress/zlib"
 	"context"
+	"encoding/base64"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"os"
-	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/elastic/apm-server/internal/beater/config"
 	"github.com/elastic/apm-server/internal/elasticsearch"
-	"github.com/elastic/apm-server/internal/version"
 	"github.com/elastic/elastic-agent-libs/monitoring"
 )
 
-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) {
-			requestPaths = append(requestPaths, r.URL.Path)
-		}))
-		defer srv.Close()
-
-		cfg := config.DefaultConfig()
-		cfg.RumConfig.Enabled = true
-		cfg.RumConfig.SourceMapping.ESConfig.Hosts = []string{srv.URL}
-		if indexPattern != "" {
-			cfg.RumConfig.SourceMapping.IndexPattern = indexPattern
-		}
-
-		fetcher, err := newSourcemapFetcher(
-			cfg.RumConfig.SourceMapping, nil,
-			nil, elasticsearch.NewClient,
-		)
-		require.NoError(t, err)
-		fetcher.Fetch(context.Background(), "name", "version", "path")
-		require.Len(t, requestPaths, 1)
-
-		path := requestPaths[0]
-		path = strings.TrimPrefix(path, "/")
-		path = strings.TrimSuffix(path, "/_search")
-		assert.Equal(t, expected, path)
-	}
-	t.Run("default-pattern", func(t *testing.T) {
-		test(t, "", "apm-*-sourcemap*")
-	})
-	t.Run("with-observer-version", func(t *testing.T) {
-		test(t, "blah-%{[observer.version]}-blah", fmt.Sprintf("blah-%s-blah", version.Version))
-	})
-}
-
 var validSourcemap, _ = os.ReadFile("../../testdata/sourcemap/bundle.js.map")
 
 func TestStoreUsesRUMElasticsearchConfig(t *testing.T) {
 	var called bool
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		called = true
 		w.Header().Set("X-Elastic-Product", "Elasticsearch")
-		w.Write(validSourcemap)
+		switch r.URL.Path {
+		case "/.apm-source-map":
+			// ping request from the metadata fetcher.
+			// Send a status ok
+			w.WriteHeader(http.StatusOK)
+		case "/.apm-source-map/_search":
+			// search request from the metadata fetcher
+			m := sourcemapSearchResponseBody("app", "1.0", "/bundle/path")
+			w.Write(m)
+		case "/.apm-source-map/_doc/app-1.0-/bundle/path":
+			m := sourcemapGetResponseBody(true, validSourcemap)
+			w.Write(m)
+			called = true
+		default:
+			w.WriteHeader(http.StatusTeapot)
+			t.Fatalf("unhandled request path: %s", r.URL.Path)
+		}
 	}))
 	defer ts.Close()
 
 	cfg := config.DefaultConfig()
 	cfg.RumConfig.Enabled = true
+	cfg.Kibana.Enabled = false
 	cfg.RumConfig.SourceMapping.Enabled = true
 	cfg.RumConfig.SourceMapping.ESConfig = elasticsearch.DefaultConfig()
 	cfg.RumConfig.SourceMapping.ESConfig.Hosts = []string{ts.URL}
 
-	fetcher, err := newSourcemapFetcher(
-		cfg.RumConfig.SourceMapping, nil,
+	fetcher, cancel, err := newSourcemapFetcher(
+		cfg.RumConfig.SourceMapping,
 		nil, elasticsearch.NewClient,
 	)
 	require.NoError(t, err)
+	defer cancel()
 	// 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")
+	c, err := fetcher.Fetch(context.Background(), "app", "1.0", "/bundle/path")
 	require.NoError(t, err)
+	require.NotNil(t, c)
 
 	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()
+func sourcemapSearchResponseBody(name string, version string, bundlePath string) []byte {
+	result := map[string]interface{}{
+		"hits": map[string]interface{}{
+			"total": map[string]interface{}{
+				"value": 1,
+			},
+			"hits": []map[string]interface{}{
+				{
+					"_source": map[string]interface{}{
+						"service": map[string]interface{}{
+							"name":    name,
+							"version": version,
+						},
+						"file": map[string]interface{}{
+							"path": bundlePath,
+						},
+					},
+				},
+			},
+		},
+	}
 
-	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,
+	data, err := json.Marshal(result)
+	if err != nil {
+		panic(err)
 	}
+	return data
+}
 
-	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)
+func sourcemapGetResponseBody(found bool, b []byte) []byte {
+	result := map[string]interface{}{
+		"found": found,
+		"_source": map[string]interface{}{
+			"content": encodeSourcemap(b),
+		},
+	}
 
-	assert.True(t, called)
+	data, err := json.Marshal(result)
+	if err != nil {
+		panic(err)
+	}
+	return data
+}
+
+func encodeSourcemap(sourcemap []byte) string {
+	b := &bytes.Buffer{}
+
+	z := zlib.NewWriter(b)
+	z.Write(sourcemap)
+	z.Close()
+
+	return base64.StdEncoding.EncodeToString(b.Bytes())
 }
 
 func TestQueryClusterUUIDRegistriesExist(t *testing.T) {