[Heartbeat] Dedupe screenshots / Extra Args

CHANGELOG.next.asciidoc

@@ -106,6 +106,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Change logging in logs input to structure logging. Some log message formats have changed. {pull}25299[25299]
 
 *Heartbeat*
+- Add support for screenshot blocks and use newer synthetics flags that only worker in newer synthetics betas. {pull}25808[25808]
 
 *Journalbeat*
 

heartbeat/_meta/fields.common.yml

@@ -174,6 +174,22 @@
               type: text
             - name: stack
               type: text
+        - name: screenshot_ref
+          type: group
+          dynamic: false
+          fields:
+            - name: width
+              type: integer
+              description: Width of the full screenshot in pixels.
+            - name: height
+              type: integer
+              description: Height of the full screenshot in pixels
+            - name: blocks
+              type: group
+              fields: 
+                - name: hash
+                  type: keyword
+                  description: Hash that uniquely identifies this image by content. Corresponds to block document id.
         - name: browser
           type: group
           fields:

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

@@ -5,18 +5,14 @@
 package browser
 
 import (
-	"context"
 	"fmt"
 	"os"
 	"os/user"
 	"sync"
 
-	"github.com/elastic/beats/v7/heartbeat/monitors/jobs"
 	"github.com/elastic/beats/v7/heartbeat/monitors/plugin"
-	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/logp"
-	"github.com/elastic/beats/v7/x-pack/heartbeat/monitors/browser/synthexec"
 )
 
 func init() {
@@ -25,14 +21,14 @@ func init() {
 
 var showExperimentalOnce = sync.Once{}
 
-var NotSyntheticsCapableError = fmt.Errorf("synthetic monitors cannot be created outside the official elastic docker image")
+var ErrNotSyntheticsCapableError = fmt.Errorf("synthetic monitors cannot be created outside the official elastic docker image")
 
 func create(name string, cfg *common.Config) (p plugin.Plugin, err error) {
 	// We don't want users running synthetics in environments that don't have the required GUI libraries etc, so we check
 	// this flag. When we're ready to support the many possible configurations of systems outside the docker environment
 	// we can remove this check.
 	if os.Getenv("ELASTIC_SYNTHETICS_CAPABLE") != "true" {
-		return plugin.Plugin{}, NotSyntheticsCapableError
+		return plugin.Plugin{}, ErrNotSyntheticsCapableError
 	}
 
 	showExperimentalOnce.Do(func() {
@@ -47,36 +43,10 @@ func create(name string, cfg *common.Config) (p plugin.Plugin, err error) {
 		return plugin.Plugin{}, fmt.Errorf("script monitors cannot be run as root! Current UID is %s", curUser.Uid)
 	}
 
-	ss, err := NewSuite(cfg)
+	s, err := NewSuite(cfg)
 	if err != nil {
 		return plugin.Plugin{}, err
 	}
 
-	extraArgs := []string{}
-	if ss.suiteCfg.Sandbox {
-		extraArgs = append(extraArgs, "--sandbox")
-	}
-
-	var j jobs.Job
-	if src, ok := ss.InlineSource(); ok {
-		j = synthexec.InlineJourneyJob(context.TODO(), src, ss.Params(), extraArgs...)
-	} else {
-		j = func(event *beat.Event) ([]jobs.Job, error) {
-			err := ss.Fetch()
-			if err != nil {
-				return nil, fmt.Errorf("could not fetch for suite job: %w", err)
-			}
-			sj, err := synthexec.SuiteJob(context.TODO(), ss.Workdir(), ss.Params(), extraArgs...)
-			if err != nil {
-				return nil, err
-			}
-			return sj(event)
-		}
-	}
-
-	return plugin.Plugin{
-		Jobs:      []jobs.Job{j},
-		Close:     ss.Close,
-		Endpoints: 1,
-	}, nil
+	return s.plugin(), nil
 }

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

@@ -25,8 +25,9 @@ type Config struct {
 	// Name is optional for lightweight checks but required for browsers
 	Name string `config:"name"`
 	// Id is optional for lightweight checks but required for browsers
-	Id      string `config:"id"`
-	Sandbox bool   `config:"sandbox"`
+	Id             string   `config:"id"`
+	Sandbox        bool     `config:"sandbox"`
+	SyntheticsArgs []string `config:"synthetics_args"`
 }
 
 var ErrNameRequired = fmt.Errorf("config 'name' must be specified for this monitor")

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

@@ -0,0 +1,108 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package browser
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/elastic/beats/v7/heartbeat/monitors/jobs"
+	"github.com/elastic/beats/v7/heartbeat/monitors/plugin"
+	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/x-pack/heartbeat/monitors/browser/synthexec"
+)
+
+type JourneyLister func(ctx context.Context, suitePath string, params common.MapStr) (journeyNames []string, err error)
+
+type Suite struct {
+	rawCfg   *common.Config
+	suiteCfg *Config
+}
+
+func NewSuite(rawCfg *common.Config) (*Suite, error) {
+	s := &Suite{
+		rawCfg:   rawCfg,
+		suiteCfg: DefaultConfig(),
+	}
+	err := rawCfg.Unpack(s.suiteCfg)
+	if err != nil {
+		return nil, ErrBadConfig(err)
+	}
+
+	return s, nil
+}
+
+func ErrBadConfig(err error) error {
+	return fmt.Errorf("could not parse suite config: %w", err)
+}
+
+func (s *Suite) String() string {
+	panic("implement me")
+}
+
+func (s *Suite) Fetch() error {
+	return s.suiteCfg.Source.Active().Fetch()
+}
+
+func (s *Suite) Workdir() string {
+	return s.suiteCfg.Source.Active().Workdir()
+}
+
+func (s *Suite) InlineSource() (string, bool) {
+	if s.suiteCfg.Source.Inline != nil {
+		return s.suiteCfg.Source.Inline.Script, true
+	}
+	return "", false
+}
+
+func (s *Suite) Params() map[string]interface{} {
+	return s.suiteCfg.Params
+}
+
+func (s *Suite) Close() error {
+	if s.suiteCfg.Source.ActiveMemo != nil {
+		s.suiteCfg.Source.ActiveMemo.Close()
+	}
+
+	return nil
+}
+
+func (s *Suite) extraArgs() []string {
+	extraArgs := s.suiteCfg.SyntheticsArgs
+	if s.suiteCfg.Sandbox {
+		extraArgs = append(extraArgs, "--sandbox")
+	}
+
+	return extraArgs
+}
+
+func (s *Suite) jobs() []jobs.Job {
+	var j jobs.Job
+	if src, ok := s.InlineSource(); ok {
+		j = synthexec.InlineJourneyJob(context.TODO(), src, s.Params(), s.extraArgs()...)
+	} else {
+		j = func(event *beat.Event) ([]jobs.Job, error) {
+			err := s.Fetch()
+			if err != nil {
+				return nil, fmt.Errorf("could not fetch for suite job: %w", err)
+			}
+			sj, err := synthexec.SuiteJob(context.TODO(), s.Workdir(), s.Params(), s.extraArgs()...)
+			if err != nil {
+				return nil, err
+			}
+			return sj(event)
+		}
+	}
+	return []jobs.Job{j}
+}
+
+func (s *Suite) plugin() plugin.Plugin {
+	return plugin.Plugin{
+		Jobs:      s.jobs(),
+		Close:     s.Close,
+		Endpoints: 1,
+	}
+}

x-pack/heartbeat/monitors/browser/suite_runner_test.go → x-pack/heartbeat/monitors/browser/suite_test.go

@@ -7,6 +7,7 @@ package browser
 import (
 	"path"
 	"path/filepath"
+	"reflect"
 	"runtime"
 	"testing"
 
@@ -114,3 +115,37 @@ func TestEmptySource(t *testing.T) {
 	require.Regexp(t, ErrBadConfig(source.ErrInvalidSource), e)
 	require.Nil(t, s)
 }
+
+func TestExtraArgs(t *testing.T) {
+	tests := []struct {
+		name string
+		cfg  *Config
+		want []string
+	}{
+		{
+			"no args",
+			&Config{},
+			nil,
+		},
+		{
+			"sandbox",
+			&Config{Sandbox: true},
+			[]string{"--sandbox"},
+		},
+		{
+			"kitchen sink",
+			&Config{SyntheticsArgs: []string{"--capability", "trace"}, Sandbox: true},
+			[]string{"--capability", "trace", "--sandbox"},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &Suite{
+				suiteCfg: tt.cfg,
+			}
+			if got := s.extraArgs(); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("Suite.extraArgs() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

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

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/elastic/beats/v7/libbeat/beat/events"
 	"github.com/elastic/beats/v7/libbeat/processors/add_data_stream_index"
 
 	"github.com/gofrs/uuid"
@@ -108,11 +109,21 @@ func (je *journeyEnricher) enrichSynthEvent(event *beat.Event, se *SynthEvent) e
 	case "step/end":
 		je.stepCount++
 	case "step/screenshot":
+		fallthrough
+	case "step/screenshot_ref":
+		fallthrough
+	case "screenshot/block":
 		add_data_stream_index.SetEventDataset(event, "browser_screenshot")
 	case "journey/network_info":
 		add_data_stream_index.SetEventDataset(event, "browser_network")
 	}
 
+	if se.Id != "" {
+		event.SetID(se.Id)
+		// This is only relevant for screenshots, which have a specific ID
+		// In that case we always want to issue an update op
+		event.Meta.Put(events.FieldMetaOpType, events.OpTypeCreate)
+	}
 	eventext.MergeEventFields(event, se.ToMap())
 
 	if je.urlFields == nil {