Fix double fetch (#35843) (#35889)

Fixes #35646 by only unpacking project monitors once.

This fixes the ever growing temp folder issue and is more efficient to boot. Previously we would call fetch on a monitor source every time it was run, but only cleanup the fetched resource once, when the monitor was unloaded. We now fetch once and cleanup once.

This project also fixes the very confusing issue of two files browser/project.go and browser/source/project.go, we have renamed browser/project.go to browser/sourcejob.go which makes reasoning about this change simpler.

(cherry picked from commit f7111dc)

Co-authored-by: Andrew Cholakian <andrewvc@elastic.co>

CHANGELOG.next.asciidoc

@@ -161,6 +161,7 @@ automatic splitting at root level, if root level element is an array. {pull}3415
 - Fix formatting issue with socket trace timeout. {pull}35434[35434]
 - Update gval version. {pull}35636[35636]
 - Fix serialization of processors when running diagnostics. {pull}35698[35698]
+- Fix temp dir running out of space with project monitors. {issue}35843[35843]
 
 *Heartbeat*
 

x-pack/heartbeat/monitors/browser/browser.go

@@ -34,7 +34,7 @@ func create(name string, cfg *config.C) (p plugin.Plugin, err error) {
  return plugin.Plugin{}, fmt.Errorf("script monitors cannot be run as root")
  }
 
- s, err := NewProject(cfg)
+ s, err := NewSourceJob(cfg)
  if err != nil {
  return plugin.Plugin{}, err
  }

x-pack/heartbeat/monitors/browser/source/project.go

@@ -17,6 +17,7 @@ import (
  "path/filepath"
  "regexp"
  "strings"
+ "sync"
  "syscall"
 
  "github.com/elastic/elastic-agent-libs/logp"
@@ -26,6 +27,8 @@ import (
 type ProjectSource struct {
  Content string `config:"content" json:"content"`
  TargetDirectory string
+ fetched bool
+ mtx sync.Mutex
 }
 
 var ErrNoContent = fmt.Errorf("no 'content' value specified for project monitor source")
@@ -39,6 +42,14 @@ func (p *ProjectSource) Validate() error {
 }
 
 func (p *ProjectSource) Fetch() error {
+ // We only need to unzip the source exactly once
+ p.mtx.Lock()
+ defer p.mtx.Unlock()
+ if p.fetched {
+ logp.L().Debugf("browser project: re-use already unpacked source: %s", p.Workdir())
+ return nil
+ }
+
  decodedBytes, err := base64.StdEncoding.DecodeString(p.Content)
  if err != nil {
  return err
@@ -60,6 +71,9 @@ func (p *ProjectSource) Fetch() error {
  if err != nil {
  return fmt.Errorf("could not make temp dir for unzipping project source: %w", err)
  }
+
+ logp.L().Debugf("browser project: unpack source: %s", p.Workdir())
+
  err = os.Chmod(p.TargetDirectory, defaultMod)
  if err != nil {
  return fmt.Errorf("failed assigning default mode %s to temp dir: %w", defaultMod, err)
@@ -81,6 +95,8 @@ func (p *ProjectSource) Fetch() error {
  }
  }
 
+ // We've succeeded, mark the fetch as a success
+ p.fetched = true
  return nil
 }
 
@@ -142,6 +158,8 @@ func (p *ProjectSource) Workdir() string {
 }
 
 func (p *ProjectSource) Close() error {
+ logp.L().Debugf("browser project: close project source: %s", p.Workdir())
+
  if p.TargetDirectory != "" {
  return os.RemoveAll(p.TargetDirectory)
  }

x-pack/heartbeat/monitors/browser/source/project_test.go

@@ -50,11 +50,29 @@ func TestProjectSource(t *testing.T) {
  return
  }
  require.NoError(t, err)
+
  fetchAndValidate(t, psrc)
  })
  }
 }
 
+func TestFetchCaching(t *testing.T) {
+ cfg := mapstr.M{
+ "content": "UEsDBBQACAAIAJ27qVQAAAAAAAAAAAAAAAAiAAAAZXhhbXBsZXMvdG9kb3MvYWR2YW5jZWQuam91cm5leS50c5VRPW/CMBDd+RWnLA0Sigt0KqJqpbZTN+iEGKzkIC6JbfkuiBTx3+uEEAGlgi7Rnf38viIESCLkR/FJ6Eis1VIjpanATBKrWFCpOUU/kcCNzG2GJNgkhoRM1lLHmERfpnAay4ipo3JrHMMWmjPYwcKZHILn33zBqIV3ADIjkxdrJ4y251eZJFNJq3b1Hh1XJx+KeKK+8XATpxiv3o07RidI7Ex5OOocTEQixcz6mF66MRgGXkmxMhqkTiA2VcJ6NQsgpZcZAnueoAfhFqxcYs9/ncwJdl0YP9XeY6OJgb3qFDcMYwhejb5jsAUDyYxBaSi9HmCJlfZJ2vCYNCpc1h2d5m8AB/r99cU+GmS/hpwXc4nmrKh/K917yK57VqZe1lU6zM26WvIiY2WbHunWIiusb3IWVBP0/bP9NGinYTC/qcqWLloY9ybjNAy5VbzYdP1sdz3+8FqJleqsP7/ONPjjp++TPgS3eaks/wBQSwcIVYIEHGwBAADRAwAAUEsDBBQACAAIAJ27qVQAAAAAAAAAAAAAAAAZAAAAZXhhbXBsZXMvdG9kb3MvaGVscGVycy50c5VUTYvbMBC9768YRGAVyKb0uktCu9CeektvpRCtM4nFKpKQxt2kwf+9I9lJ5cRb6MWW5+u9eTOW3nsXCE4QCf0M8OCxImhhG9wexCc0KpKuPsSjpRr5FMXTXeVsJDBObT57v+I8WID0aoczaIKZwmIJpzvIFaUwqrFVDcp7MQPFdSqQlxAA9aY0QUqe7xw5mQo8saflZ3uGUpvNdxVfh1DEliHWmuOyGSan9GrXY4hdSW19Q1yswJ9Ika1zi28P5DZOZCZnjp2Pjh5lhr71+YAxSvHFEgZx20UqGVdoWGAXGFo0Zp5sD0YnOXX+uMi71TY3nTh2PYy0HZCaYMsm0umrC2cYuWYpStwWlksgPNBC9CKJ9UDqGDFQAv7GrFb6N/aqD0hEtl9pX9VYvQLViroR5KZqFXmlVEXmyDNJWS0wkT1aiqPD6fZPynIsEznoYDqdG7Q7qqcs2DPKzOVG7EyHhSj25n0Zyw62PJvcwH2vzz1PN3czSrifwHlaZfUbThuMFNzxPyj1GVeE/rHWRr2guaz1e6wu0foSmhPTL3DwiuqFshVDu/D4aPSPjz/FIK1n9dwQOfu3gk7pL9k4jK+M5lk0LBRy9CB7nn2yD+cStfuFQQ5+riK9kJQ3JV9cbCmuh1n6HF3h5LleimS7GkoynWVL5+KWS6h/AFBLBwgvDHpj+wEAAC8FAABQSwECLQMUAAgACACdu6lUVYIEHGwBAADRAwAAIgAAAAAAAAAAACAApIEAAAAAZXhhbXBsZXMvdG9kb3MvYWR2YW5jZWQuam91cm5leS50c1BLAQItAxQACAAIAJ27qVQvDHpj+wEAAC8FAAAZAAAAAAAAAAAAIACkgbwBAABleGFtcGxlcy90b2Rvcy9oZWxwZXJzLnRzUEsFBgAAAAACAAIAlwAAAP4DAAAAAA==",
+ }
+ psrc, err := dummyPSource(cfg)
+ require.NoError(t, err)
+ defer psrc.Close()
+
+ err = psrc.Fetch()
+ require.NoError(t, err)
+ wdir := psrc.Workdir()
+ err = psrc.Fetch()
+ require.NoError(t, err)
+ wdirNext := psrc.Workdir()
+ require.Equal(t, wdir, wdirNext)
+}
+
 func validateFileContents(t *testing.T, dir string) {
  expected := []string{
  "examples/todos/helpers.ts",
@@ -73,6 +91,9 @@ func validateFileContents(t *testing.T, dir string) {
 }
 
 func fetchAndValidate(t *testing.T, psrc *ProjectSource) {
+ defer func() {
+ _ = psrc.Close()
+ }()
  err := psrc.Fetch()
  require.NoError(t, err)
 

x-pack/heartbeat/monitors/browser/project.go → x-pack/heartbeat/monitors/browser/sourcejob.go

@@ -18,30 +18,27 @@ import (
  "github.com/elastic/beats/v7/x-pack/heartbeat/monitors/browser/synthexec"
  "github.com/elastic/elastic-agent-libs/config"
  "github.com/elastic/elastic-agent-libs/logp"
- "github.com/elastic/elastic-agent-libs/mapstr"
 )
 
-type JourneyLister func(ctx context.Context, projectPath string, params mapstr.M) (journeyNames []string, err error)
-
-type Project struct {
+type SourceJob struct {
  rawCfg *config.C
- projectCfg *Config
+ browserCfg *Config
  ctx context.Context
  cancel context.CancelFunc
 }
 
-func NewProject(rawCfg *config.C) (*Project, error) {
- // Global project context to cancel all jobs
+func NewSourceJob(rawCfg *config.C) (*SourceJob, error) {
+ // Global browser context to cancel all jobs
  // on close
  ctx, cancel := context.WithCancel(context.Background())
 
- s := &Project{
+ s := &SourceJob{
  rawCfg: rawCfg,
- projectCfg: DefaultConfig(),
+ browserCfg: DefaultConfig(),
  ctx: ctx,
  cancel: cancel,
  }
- err := rawCfg.Unpack(s.projectCfg)
+ err := rawCfg.Unpack(s.browserCfg)
  if err != nil {
  return nil, ErrBadConfig(err)
  }
@@ -50,31 +47,31 @@ func NewProject(rawCfg *config.C) (*Project, error) {
 }
 
 func ErrBadConfig(err error) error {
- return fmt.Errorf("could not parse project config: %w", err)
+ return fmt.Errorf("could not parse browser config: %w", err)
 }
 
-func (p *Project) String() string {
+func (sj *SourceJob) String() string {
  panic("implement me")
 }
 
-func (p *Project) Fetch() error {
- return p.projectCfg.Source.Active().Fetch()
+func (sj *SourceJob) Fetch() error {
+ return sj.browserCfg.Source.Active().Fetch()
 }
 
-func (p *Project) Workdir() string {
- return p.projectCfg.Source.Active().Workdir()
+func (sj *SourceJob) Workdir() string {
+ return sj.browserCfg.Source.Active().Workdir()
 }
 
-func (p *Project) Params() map[string]interface{} {
- return p.projectCfg.Params
+func (sj *SourceJob) Params() map[string]interface{} {
+ return sj.browserCfg.Params
 }
 
-func (p *Project) FilterJourneys() synthexec.FilterJourneyConfig {
- return p.projectCfg.FilterJourneys
+func (sj *SourceJob) FilterJourneys() synthexec.FilterJourneyConfig {
+ return sj.browserCfg.FilterJourneys
 }
 
-func (p *Project) StdFields() stdfields.StdMonitorFields {
- sFields, err := stdfields.ConfigToStdMonitorFields(p.rawCfg)
+func (sj *SourceJob) StdFields() stdfields.StdMonitorFields {
+ sFields, err := stdfields.ConfigToStdMonitorFields(sj.rawCfg)
  // Should be impossible since outer monitor.go should run this same code elsewhere
  // TODO: Just pass stdfields in to remove second deserialize
  if err != nil {
@@ -83,45 +80,45 @@ func (p *Project) StdFields() stdfields.StdMonitorFields {
  return sFields
 }
 
-func (p *Project) Close() error {
- if p.projectCfg.Source.ActiveMemo != nil {
- p.projectCfg.Source.ActiveMemo.Close()
+func (sj *SourceJob) Close() error {
+ if sj.browserCfg.Source.ActiveMemo != nil {
+ sj.browserCfg.Source.ActiveMemo.Close()
  }
 
  // Cancel running jobs ctxs
- p.cancel()
+ sj.cancel()
 
  return nil
 }
 
-func (p *Project) extraArgs() []string {
- extraArgs := p.projectCfg.SyntheticsArgs
- if len(p.projectCfg.PlaywrightOpts) > 0 {
- s, err := json.Marshal(p.projectCfg.PlaywrightOpts)
+func (sj *SourceJob) extraArgs() []string {
+ extraArgs := sj.browserCfg.SyntheticsArgs
+ if len(sj.browserCfg.PlaywrightOpts) > 0 {
+ s, err := json.Marshal(sj.browserCfg.PlaywrightOpts)
  if err != nil {
  // This should never happen, if it was parsed as a config it should be serializable
- logp.L().Warn("could not serialize playwright options '%v': %w", p.projectCfg.PlaywrightOpts, err)
+ logp.L().Warn("could not serialize playwright options '%v': %w", sj.browserCfg.PlaywrightOpts, err)
  } else {
  extraArgs = append(extraArgs, "--playwright-options", string(s))
  }
  }
- if p.projectCfg.IgnoreHTTPSErrors {
+ if sj.browserCfg.IgnoreHTTPSErrors {
  extraArgs = append(extraArgs, "--ignore-https-errors")
  }
- if p.projectCfg.Sandbox {
+ if sj.browserCfg.Sandbox {
  extraArgs = append(extraArgs, "--sandbox")
  }
- if p.projectCfg.Screenshots != "" {
- extraArgs = append(extraArgs, "--screenshots", p.projectCfg.Screenshots)
+ if sj.browserCfg.Screenshots != "" {
+ extraArgs = append(extraArgs, "--screenshots", sj.browserCfg.Screenshots)
  }
- if p.projectCfg.Throttling != nil {
- switch t := p.projectCfg.Throttling.(type) {
+ if sj.browserCfg.Throttling != nil {
+ switch t := sj.browserCfg.Throttling.(type) {
  case bool:
  if !t {
  extraArgs = append(extraArgs, "--no-throttling")
  }
  case string:
- extraArgs = append(extraArgs, "--throttling", fmt.Sprintf("%v", p.projectCfg.Throttling))
+ extraArgs = append(extraArgs, "--throttling", fmt.Sprintf("%v", sj.browserCfg.Throttling))
  case map[string]interface{}:
  j, err := json.Marshal(t)
  if err != nil {
@@ -135,22 +132,22 @@ func (p *Project) extraArgs() []string {
  return extraArgs
 }
 
-func (p *Project) jobs() []jobs.Job {
+func (sj *SourceJob) jobs() []jobs.Job {
  var j jobs.Job
 
- isScript := p.projectCfg.Source.Inline != nil
- ctx := context.WithValue(p.ctx, synthexec.SynthexecTimeout, p.projectCfg.Timeout+30*time.Second)
+ isScript := sj.browserCfg.Source.Inline != nil
+ ctx := context.WithValue(sj.ctx, synthexec.SynthexecTimeout, sj.browserCfg.Timeout+30*time.Second)
 
  if isScript {
- src := p.projectCfg.Source.Inline.Script
- j = synthexec.InlineJourneyJob(ctx, src, p.Params(), p.StdFields(), p.extraArgs()...)
+ src := sj.browserCfg.Source.Inline.Script
+ j = synthexec.InlineJourneyJob(ctx, src, sj.Params(), sj.StdFields(), sj.extraArgs()...)
  } else {
  j = func(event *beat.Event) ([]jobs.Job, error) {
- err := p.Fetch()
+ err := sj.Fetch()
  if err != nil {
- return nil, fmt.Errorf("could not fetch for project job: %w", err)
+ return nil, fmt.Errorf("could not fetch for browser source job: %w", err)
  }
- sj, err := synthexec.ProjectJob(ctx, p.Workdir(), p.Params(), p.FilterJourneys(), p.StdFields(), p.extraArgs()...)
+ sj, err := synthexec.ProjectJob(ctx, sj.Workdir(), sj.Params(), sj.FilterJourneys(), sj.StdFields(), sj.extraArgs()...)
  if err != nil {
  return nil, err
  }
@@ -160,10 +157,10 @@ func (p *Project) jobs() []jobs.Job {
  return []jobs.Job{j}
 }
 
-func (p *Project) plugin() plugin.Plugin {
+func (sj *SourceJob) plugin() plugin.Plugin {
  return plugin.Plugin{
- Jobs: p.jobs(),
- DoClose: p.Close,
+ Jobs: sj.jobs(),
+ DoClose: sj.Close,
  Endpoints: 1,
  }
 }

x-pack/heartbeat/monitors/browser/project_test.go → x-pack/heartbeat/monitors/browser/sourcejob_test.go

@@ -44,7 +44,7 @@ func TestValidLocal(t *testing.T) {
  },
  "timeout": timeout,
  })
- _, e := NewProject(cfg)
+ _, e := NewSourceJob(cfg)
  require.Error(t, e)
 }
 
@@ -66,10 +66,10 @@ func TestValidInline(t *testing.T) {
  },
  "timeout": timeout,
  })
- s, e := NewProject(cfg)
+ s, e := NewSourceJob(cfg)
  require.NoError(t, e)
  require.NotNil(t, s)
- require.Equal(t, script, s.projectCfg.Source.Inline.Script)
+ require.Equal(t, script, s.browserCfg.Source.Inline.Script)
  require.Equal(t, "", s.Workdir())
  require.Equal(t, testParams, s.Params())
 
@@ -86,7 +86,7 @@ func TestNameRequired(t *testing.T) {
  },
  },
  })
- _, e := NewProject(cfg)
+ _, e := NewSourceJob(cfg)
  require.Regexp(t, ErrNameRequired, e)
 }
 
@@ -99,15 +99,15 @@ func TestIDRequired(t *testing.T) {
  },
  },
  })
- _, e := NewProject(cfg)
+ _, e := NewSourceJob(cfg)
  require.Regexp(t, ErrIdRequired, e)
 }
 
 func TestEmptySource(t *testing.T) {
  cfg := conf.MustNewConfigFrom(mapstr.M{
  "source": mapstr.M{},
  })
- s, e := NewProject(cfg)
+ s, e := NewSourceJob(cfg)
 
  require.Regexp(t, ErrBadConfig(source.ErrInvalidSource), e)
  require.Nil(t, s)
@@ -196,8 +196,8 @@ func TestExtraArgs(t *testing.T) {
  }
  for _, tt := range tests {
  t.Run(tt.name, func(t *testing.T) {
- s := &Project{
- projectCfg: tt.cfg,
+ s := &SourceJob{
+ browserCfg: tt.cfg,
  }
  if got := s.extraArgs(); !reflect.DeepEqual(got, tt.want) {
  t.Errorf("Project.extraArgs() = %v, want %v", got, tt.want)
@@ -217,9 +217,9 @@ func TestEmptyTimeout(t *testing.T) {
  },
  },
  })
- s, e := NewProject(cfg)
+ s, e := NewSourceJob(cfg)
 
  require.NoError(t, e)
  require.NotNil(t, s)
- require.Equal(t, s.projectCfg.Timeout, defaults.Timeout)
+ require.Equal(t, s.browserCfg.Timeout, defaults.Timeout)
 }