feat(dsl): implement progress (#13)

Noticed while working on ooni/probe#2502

pkg/dsl/ast.go

@@ -112,6 +112,9 @@ func NewASTLoader() *ASTLoader {
 	// parallel.go
 	al.RegisterCustomLoaderRule(&runStagesInParallelLoader{})
 
+	// progress.go
+	al.RegisterCustomLoaderRule(&wrapWithProgressLoader{})
+
 	// quichandshake.go
 	al.RegisterCustomLoaderRule(&quicHandshakeLoader{})
 

pkg/dsl/example_test.go

@@ -111,7 +111,7 @@ func Example_internalDSL() {
 
 	// Create a measurement runtime using measurexlite as the underlying
 	// measurement library such that we also collect observations.
-	rtx := dsl.NewMeasurexliteRuntime(log.Log, &dsl.NullMetrics{}, time.Now())
+	rtx := dsl.NewMeasurexliteRuntime(log.Log, &dsl.NullMetrics{}, &dsl.NullProgressMeter{}, time.Now())
 
 	// Create the void input for the pipeline.
 	input := dsl.NewValue(&dsl.Void{})
@@ -166,7 +166,7 @@ func Example_externalDSL() {
 
 	// Create a measurement runtime using measurexlite as the underlying
 	// measurement library such that we also collect observations.
-	rtx := dsl.NewMeasurexliteRuntime(log.Log, &dsl.NullMetrics{}, time.Now())
+	rtx := dsl.NewMeasurexliteRuntime(log.Log, &dsl.NullMetrics{}, &dsl.NullProgressMeter{}, time.Now())
 
 	// Create the void input for the pipeline. We need to cast the input to
 	// a generic Maybe because there's dynamic type checking when running an
@@ -211,7 +211,7 @@ func Example_singleEndpointInternalDSL() {
 	metrics := dsl.NewAccountingMetrics()
 
 	// create a measurement runtime
-	rtx := dsl.NewMeasurexliteRuntime(log.Log, metrics, time.Now())
+	rtx := dsl.NewMeasurexliteRuntime(log.Log, metrics, &dsl.NullProgressMeter{}, time.Now())
 
 	// run the measurement pipeline
 	_ = pipeline.Run(context.Background(), rtx, dsl.NewValue(&dsl.Void{}))
@@ -256,7 +256,7 @@ func Example_singleEndpointExternalDSL() {
 	metrics := dsl.NewAccountingMetrics()
 
 	// create a measurement runtime
-	rtx := dsl.NewMeasurexliteRuntime(log.Log, metrics, time.Now())
+	rtx := dsl.NewMeasurexliteRuntime(log.Log, metrics, &dsl.NullProgressMeter{}, time.Now())
 
 	// run the measurement pipeline
 	_ = runnable.Run(context.Background(), rtx, dsl.NewValue(&dsl.Void{}).AsGeneric())

pkg/dsl/measurexlite.go

@@ -17,6 +17,9 @@ type MeasurexliteRuntime struct {
 	// metrics contains the metrics.
 	metrics Metrics
 
+	// progress is the ProgressMeter to use.
+	progress ProgressMeter
+
 	// runtime is the MinimalRuntime we compose with.
 	runtime *MinimalRuntime
 
@@ -25,9 +28,15 @@ type MeasurexliteRuntime struct {
 }
 
 // NewMeasurexliteRuntime creates a new [MeasurexliteRuntime].
-func NewMeasurexliteRuntime(logger model.Logger, metrics Metrics, zeroTime time.Time) *MeasurexliteRuntime {
+func NewMeasurexliteRuntime(
+	logger model.Logger,
+	metrics Metrics,
+	progress ProgressMeter,
+	zeroTime time.Time,
+) *MeasurexliteRuntime {
 	return &MeasurexliteRuntime{
 		metrics:  metrics,
+		progress: progress,
 		runtime:  NewMinimalRuntime(logger),
 		zeroTime: zeroTime,
 	}
@@ -40,6 +49,11 @@ func (r *MeasurexliteRuntime) Close() error {
 	return r.runtime.Close()
 }
 
+// ProgressMeter implements Runtime.
+func (r *MeasurexliteRuntime) ProgressMeter() ProgressMeter {
+	return r.progress
+}
+
 // Metrics implements Runtime.
 func (r *MeasurexliteRuntime) Metrics() Metrics {
 	return r.metrics

pkg/dsl/measurexlite_test.go

@@ -59,7 +59,8 @@ func TestMeasurexliteHTTPIncludeResponseBodySnapshot(t *testing.T) {
 	// define the function to run the measurement
 	measure := func(options ...HTTPTransactionOption) (Maybe[*HTTPResponse], *Observations) {
 		pipeline := makePipeline(options...)
-		rtx := NewMeasurexliteRuntime(model.DiscardLogger, &NullMetrics{}, time.Now())
+		meter := &NullProgressMeter{}
+		rtx := NewMeasurexliteRuntime(model.DiscardLogger, &NullMetrics{}, meter, time.Now())
 		input := NewValue(&Void{})
 		output := pipeline.Run(context.Background(), rtx, input)
 		observations := ReduceObservations(rtx.ExtractObservations()...)

pkg/dsl/progress.go

@@ -0,0 +1,130 @@
+package dsl
+
+import (
+	"context"
+	"encoding/json"
+	"sync"
+
+	"github.com/ooni/probe-engine/pkg/model"
+)
+
+// ProgressMeter tracks progress.
+type ProgressMeter interface {
+	// IncrementProgress increments the progress meter by adding the given delta
+	// to the current progress meter value. The progress meter value is a float
+	// number where 0 means beginning and 1.0 means we are done.
+	IncrementProgress(delta float64)
+}
+
+// NullProgressMeter is a [ProgressMeter] that does nothing. The zero
+// value of this struct is ready to use.
+type NullProgressMeter struct{}
+
+var _ ProgressMeter = &NullProgressMeter{}
+
+// IncrementProgress implements ProgressMeter.
+func (pm *NullProgressMeter) IncrementProgress(value float64) {
+	// nothing
+}
+
+// ProgressMeterExperimentCallbacks wraps [model.ExperimentCallbacks] and
+// implements [ProgressMeter]. The zero value is not ready to use; you should
+// construct using the [NewProgressMeterExperimentCallbacks] factory.
+type ProgressMeterExperimentCallbacks struct {
+	callbacks model.ExperimentCallbacks
+	mu        sync.Mutex
+	total     float64
+}
+
+// NewProgressMeterExperimentCallbacks constructs a new [ProgressMeterExperimentCallbacks].
+func NewProgressMeterExperimentCallbacks(cb model.ExperimentCallbacks) *ProgressMeterExperimentCallbacks {
+	return &ProgressMeterExperimentCallbacks{
+		callbacks: cb,
+		mu:        sync.Mutex{},
+		total:     0,
+	}
+}
+
+var _ ProgressMeter = &ProgressMeterExperimentCallbacks{}
+
+// IncrementProgress implements ProgressMeter.
+func (pm *ProgressMeterExperimentCallbacks) IncrementProgress(delta float64) {
+	pm.mu.Lock()
+	if delta >= 0 {
+		pm.total += delta
+		if pm.total > 1.0 {
+			pm.total = 1.0
+		}
+	}
+	total := pm.total
+	pm.mu.Unlock()
+	pm.callbacks.OnProgress(total, "")
+}
+
+// WrapWithProgress wraps a list of stages such that each stage increments the
+// progress of running a measurement by an equal contribution.
+func WrapWithProgress(input ...Stage[*Void, *Void]) (output []Stage[*Void, *Void]) {
+	var delta float64
+	if len(input) > 0 {
+		delta = 1 / float64(len(input))
+	}
+	for _, stage := range input {
+		output = append(output, &wrapWithProgressStage{delta, stage})
+	}
+	return output
+}
+
+type wrapWithProgressStage struct {
+	delta float64
+	stage Stage[*Void, *Void]
+}
+
+const wrapWithProgressStageName = "wrap_with_progress"
+
+type wrapWithProgressStageArguments struct {
+	Delta float64 `json:"delta"`
+}
+
+// ASTNode implements Stage.
+func (sx *wrapWithProgressStage) ASTNode() *SerializableASTNode {
+	// Note: we serialize the structure because this gives us forward compatibility (i.e., we
+	// may add a field to a future version without breaking the AST structure and old probes will
+	// be fine as long as the zero value of the new field is the default)
+	return &SerializableASTNode{
+		StageName: wrapWithProgressStageName,
+		Arguments: &wrapWithProgressStageArguments{sx.delta},
+		Children:  []*SerializableASTNode{sx.stage.ASTNode()},
+	}
+}
+
+type wrapWithProgressLoader struct{}
+
+// Load implements ASTLoaderRule.
+func (*wrapWithProgressLoader) Load(loader *ASTLoader, node *LoadableASTNode) (RunnableASTNode, error) {
+	var config wrapWithProgressStageArguments
+	if err := json.Unmarshal(node.Arguments, &config); err != nil {
+		return nil, err
+	}
+	runnables, err := loader.LoadChildren(node)
+	if err != nil {
+		return nil, err
+	}
+	if len(runnables) != 1 {
+		return nil, ErrInvalidNumberOfChildren
+	}
+	runnables0 := &RunnableASTNodeStage[*Void, *Void]{runnables[0]}
+	stage := &wrapWithProgressStage{config.Delta, runnables0}
+	return &StageRunnableASTNode[*Void, *Void]{stage}, nil
+}
+
+// StageName implements ASTLoaderRule.
+func (*wrapWithProgressLoader) StageName() string {
+	return wrapWithProgressStageName
+}
+
+// Run implements Stage.
+func (sx *wrapWithProgressStage) Run(ctx context.Context, rtx Runtime, input Maybe[*Void]) Maybe[*Void] {
+	output := sx.stage.Run(ctx, rtx, input)
+	rtx.ProgressMeter().IncrementProgress(sx.delta)
+	return output
+}

pkg/dsl/qa_test.go

@@ -131,7 +131,7 @@ func qaNewEnvironment() *netemx.QAEnv {
 
 func qaRunNode(metrics dsl.Metrics, runnable dsl.RunnableASTNode) (*dsl.Observations, error) {
 	input := dsl.NewValue(&dsl.Void{}).AsGeneric()
-	rtx := dsl.NewMeasurexliteRuntime(log.Log, metrics, time.Now())
+	rtx := dsl.NewMeasurexliteRuntime(log.Log, metrics, &dsl.NullProgressMeter{}, time.Now())
 	if err := dsl.Try(runnable.Run(context.Background(), rtx, input)); err != nil {
 		return nil, err
 	}

pkg/dsl/runtime.go

@@ -19,14 +19,17 @@ type Runtime interface {
 	// Close closes all the closers tracker by the runtime.
 	Close() error
 
+	// Logger returns the logger to use.
+	Logger() model.Logger
+
 	// Metrics returns the metrics to use.
 	Metrics() Metrics
 
 	// NewTrace creates a new measurement trace.
 	NewTrace() Trace
 
-	// Logger returns the logger to use.
-	Logger() model.Logger
+	// ProgressMeter returns the progress meter to use.
+	ProgressMeter() ProgressMeter
 
 	// SaveObservations saves the given observations into the runtime.
 	SaveObservations(observations ...*Observations)
@@ -82,7 +85,7 @@ func (r *MinimalRuntime) Close() error {
 	return nil
 }
 
-// ExtractObservations implements Trace.
+// ExtractObservations implements Runtime.
 func (r *MinimalRuntime) ExtractObservations() []*Observations {
 	defer r.mu.Unlock()
 	r.mu.Lock()
@@ -91,6 +94,11 @@ func (r *MinimalRuntime) ExtractObservations() []*Observations {
 	return out
 }
 
+// ProgressMeter implements Runtime.
+func (r *MinimalRuntime) ProgressMeter() ProgressMeter {
+	return &NullProgressMeter{}
+}
+
 // Metrics implements Runtime.
 func (r *MinimalRuntime) Metrics() Metrics {
 	return defaultNullMetrics

pkg/experiment/fbmessenger/dslcore.go

@@ -4,7 +4,7 @@ import "github.com/ooni/2023-05-richer-input/pkg/dsl"
 
 // DSLToplevelFunc generates the Facebook Messenger measurement pipeline.
 func DSLToplevelFunc(tk *TestKeys) dsl.Stage[*dsl.Void, *dsl.Void] {
-	return dsl.RunStagesInParallel(
+	stages := dsl.WrapWithProgress(
 
 		// stun
 		dsl.Compose4(
@@ -130,4 +130,5 @@ func DSLToplevelFunc(tk *TestKeys) dsl.Stage[*dsl.Void, *dsl.Void] {
 			),
 		),
 	)
+	return dsl.RunStagesInParallel(stages...)
 }

pkg/experiment/fbmessenger/measurer.go

@@ -58,8 +58,10 @@ func (m *Measurer) Run(ctx context.Context, args *model.ExperimentArgs) error {
 	}
 
 	// create the DSL runtime
+	meter := dsl.NewProgressMeterExperimentCallbacks(args.Callbacks)
 	rtx := dsl.NewMeasurexliteRuntime(
-		args.Session.Logger(), &dsl.NullMetrics{}, args.Measurement.MeasurementStartTimeSaved)
+		args.Session.Logger(), &dsl.NullMetrics{}, meter,
+		args.Measurement.MeasurementStartTimeSaved)
 	defer rtx.Close()
 
 	// evaluate the pipeline and handle exceptions

pkg/experiment/riseupvpn/riseupvpn.go

@@ -63,13 +63,13 @@ func (m *Measurer) Run(ctx context.Context, args *model.ExperimentArgs) error {
 	// TODO(bassosimone): both fbmessenger and riseupvpn lack
 	//
 	// 1. an explicit mechanism to report the bytes sent and received, but the
-	// implicit context-based mechanism probably works;
-	//
-	// 2. a DSL-based mechanism to increment the test progress percentage.
+	// implicit context-based mechanism probably works.
 
 	// create the DSL runtime
+	progress := dsl.NewProgressMeterExperimentCallbacks(args.Callbacks)
 	rtx := dsl.NewMeasurexliteRuntime(
-		args.Session.Logger(), &dsl.NullMetrics{}, args.Measurement.MeasurementStartTimeSaved)
+		args.Session.Logger(), &dsl.NullMetrics{}, progress,
+		args.Measurement.MeasurementStartTimeSaved)
 	defer rtx.Close()
 
 	// evaluate the pipeline and handle exceptions

pkg/ooniprobe/runner/progress.go

@@ -76,6 +76,8 @@ var _ model.ExperimentCallbacks = &progressEmitterNettest{}
 func (pe *progressEmitterNettest) OnProgress(progress float64, message string) {
 	// the view only supports setting the progress, so use the logger
 	// to make sure the message is not lost
-	pe.logger.Info(message)
+	if message != "" {
+		pe.logger.Info(message)
+	}
 	pe.view.UpdateProgressBarValueWithinRange(progress)
 }

pkg/x/cmd/riseupvpn/main.go

@@ -16,7 +16,7 @@ import (
 // isTCPGatewayAccessible returns whether a gateways is accessible.
 func isTCPGatewayAccessible(stage dsl.Stage[*dsl.Void, *dsl.Void]) bool {
 	metrics := dsl.NewAccountingMetrics()
-	rtx := dsl.NewMeasurexliteRuntime(log.Log, metrics, time.Now())
+	rtx := dsl.NewMeasurexliteRuntime(log.Log, metrics, &dsl.NullProgressMeter{}, time.Now())
 	input := dsl.NewValue(&dsl.Void{})
 	ctx := context.Background()
 	runtimex.Try0(dsl.Try(stage.Run(ctx, rtx, input)))
@@ -84,7 +84,7 @@ func mustGenerateDSL(eipService *apiEIPService, rootCA string) dsl.Stage[*dsl.Vo
 	stages = append(stages, dslRuleFetchGeoServiceURL(rootCA))
 
 	// return the composed pipeline
-	return dsl.RunStagesInParallel(stages...)
+	return dsl.RunStagesInParallel(dsl.WrapWithProgress(stages...)...)
 }
 
 func main() {