6337355964

Committed 28 Sep 2023 09:57AM UTC coverage: 81.941% (-0.003%) from 81.944%

Build # 6337355964

Build Type

push

github

Committed by

web-flow

Commit Message

refactor(scrubber): reorganize tests and expose ScrubBytes (#1320)

This set of changes is the continuation of what I started doing in
https://github.com/ooni/probe-cli/commit/4a86c5429
to revamp out scrubbing capabilities.

Reorganizing tests is pure refactoring. Exposing ScrubBytes is because I
think we can rationalize a bit what we scrub on the fly in the
`internal/model/archival.go` file and we need to indeed start scrubbing
byte arrays.

I'm doing this work as part of https://github.com/ooni/probe/issues/2531

Run Details

8 of 8 new or added lines in 4 files covered. (100.0%)

1 existing line in 1 file now uncovered.

24243 of 29586 relevant lines covered (81.94%)

36.33 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.75

/internal/memoryless/memoryless.go

// Package memoryless helps repeated calls to a function be distributed across
// time in a memoryless fashion.
package memoryless

// Adapted from https://github.com/m-lab/go/commit/df205a2a463b6624de235da6a61b409567b1ed98
// SPDX-License-Identifier: Apache-2.0

import (
        "context"
        "fmt"
        "math/rand"
        "time"
)

// Config represents the time we should wait between runs of the function.
//
// A valid config will have:
//
//        0 <= Min <= Expected <= Max (or 0 <= Min <= Expected and Max is 0)
//
// If Max is zero or unset, it will be ignored. If Min is zero or unset, it will
// be ignored.
type Config struct {
        // Expected records the expected/mean/average amount of time between runs.
        Expected time.Duration
        // Min provides clamping of the randomly produced value. All timers will wait
        // at least Min time.
        Min time.Duration
        // Max provides clamping of the randomly produced value. All timers will take
        // at most Max time.
        Max time.Duration

        // Once is provided as a helper, because frequently for unit testing and
        // integration testing, you only want the "Forever" loop to run once.
        //
        // The zero value of this struct has Once set to false, which means the value
        // only needs to be set explicitly in codepaths where it might be true.
        Once bool
}

func (c Config) waittime() time.Duration {
        wt := time.Duration(rand.ExpFloat64() * float64(c.Expected))
        if wt < c.Min {
                wt = c.Min
        }
        if c.Max != 0 && wt > c.Max {
                wt = c.Max
        }
        return wt
}

// Check whether the config contrains sensible values. It return an error if the
// config makes no mathematical sense, and nil if everything is okay.
func (c Config) Check() error {
        if !(0 <= c.Min && c.Min <= c.Expected && (c.Max == 0 || c.Expected <= c.Max)) {
                return fmt.Errorf(
                        "memoryless: the arguments to Run make no sense: it should be true that Min <= Expected <= Max (or Min <= Expected and Max is 0), "+
                                "but that is not true for Min(%v) Expected(%v) Max(%v)",
                        c.Min, c.Expected, c.Max)
        }
        return nil
}

// newTimer constructs and returns a timer. This function assumes that the
// config has no errors.
func newTimer(c Config) *time.Timer {
        return time.NewTimer(c.waittime())
}

// NewTimer constructs a single-shot time.Timer that, if repeatedly used to
// construct a series of timers, will ensure that the resulting events conform
// to the memoryless distribution. For more on how this could and should be
// used, see the comments to Ticker. It is intended to be a drop-in replacement
// for time.NewTimer.
func NewTimer(c Config) (*time.Timer, error) {
        if err := c.Check(); err != nil {
                return nil, err
        }

        return newTimer(c), nil
}

// AfterFunc constructs a single-shot time.Timer that, if repeatedly used to
// construct a series of timers, will ensure that the resulting events conform
// to the memoryless distribution. For more on how this could and should be
// used, see the comments to Ticker. It is intended to be a drop-in replacement
// for time.AfterFunc.
func AfterFunc(c Config, f func()) (*time.Timer, error) {
        if err := c.Check(); err != nil {
                return nil, err
        }

        return time.AfterFunc(c.waittime(), f), nil
}

// Ticker is a struct that waits a config.Expected amount of time on average
// between sends down the channel C. It has the same interface and requirements
// as time.Ticker. Every Ticker created must have its Stop() method called or it
// will leak a goroutine.
//
// The inter-send time is a random variable governed by the exponential
// distribution and will generate a memoryless (Poisson) distribution of channel
// reads over time, ensuring that a measurement scheme using this ticker has the
// PASTA property (Poisson Arrivals See Time Averages). This statistical
// guarantee is subject to two caveats:
//
// Caveat 1 is that, in a nod to the realities of systems needing to have
// guarantees, we allow the random wait time to be clamped both above and below.
// This means that channel events will be at least config.Min and at most
// config.Max apart in time. This clamping causes bias in the timing. For use of
// Ticker to be statistically sensible, the clamping should not be too extreme.
// The exact mathematical meaning of "too extreme" depends on your situation,
// but a nice rule of thumb is config.Min should be at most 10% of expected and
// config.Max should be at least 250% of expected. These values mean that less
// than 10% of time you will be waiting config.Min and less than 10% of the time
// you will be waiting config.Max.
//
// Caveat 2 is that this assumes that the actions performed between channel
// reads take negligible time when compared to the expected wait time.
// Memoryless sequences have the property that the times between successive
// event starts has the exponential distribution, and the exponential
// distribution can generate numbers arbitrarily close to zero (albeit
// exponentially infrequently). This code will not send on the channel if the
// other end is not ready to receive, which provides another lower bound on
// inter-event times. The only other option if the other side of the channel is
// not ready to receive would be queueing events in the channel, and that has
// some pathological cases we would like to avoid. In particular, queuing can
// cause long-term correlations if the queue gets big, which is the exact
// opposite of what a memoryless system is trying to achieve.
type Ticker struct {
        C         <-chan time.Time // The channel on which the ticks are delivered.
        config    Config
        writeChan chan<- time.Time
        cancel    func()
}

func (t *Ticker) singleIteration(ctx context.Context) {
        timer := newTimer(t.config)
        defer timer.Stop()
        // Wait until the timer is done or the context is canceled. If both conditions
        // are true, which case gets called is unspecified.
        select {
        case <-ctx.Done():
                // Please don't put code here that assumes that this code path will
                // definitely execute if the context is done. select {} doesn't promise that
                // multiple channels will get selected with equal probability, which means
                // that it could be true that the timer is done AND the context is canceled,
                // and we have no guarantee that in that case the canceled context case will
                // be the one that is selected.
        case <-timer.C:
        }
        // Just like time.Ticker, writes to the channel are non-blocking. If a user of
        // this module can't keep up with the timer they set, that's on them. There
        // are some potential pathological cases associated with queueing events in
        // the channel, and we want to avoid them.
        select {
        case t.writeChan <- time.Now():
        default:
        }
}

func (t *Ticker) runTicker(ctx context.Context) {
        // No matter what, when this function exits the channel should never be written to again.
        defer close(t.writeChan)

        if t.config.Once {
                if ctx.Err() == nil {
                        t.singleIteration(ctx)
                }
                return
        }

        // When Done() is not closed and the Deadline has not been exceeded, the error
        // is nil.
        for ctx.Err() == nil {
                t.singleIteration(ctx)
        }
}

// Stop the ticker goroutine.
func (t *Ticker) Stop() {
        t.cancel()
}

// MakeTicker is a deprecated alias for NewTicker
var MakeTicker = NewTicker

// NewTicker creates a new memoryless ticker. The returned struct is compatible
// with the time.Ticker struct interface, and everywhere you use a time.Ticker,
// you can use a memoryless.Ticker.
func NewTicker(ctx context.Context, config Config) (*Ticker, error) {
        if err := config.Check(); err != nil {
                return nil, err
        }
        c := make(chan time.Time)
        ctx, cancel := context.WithCancel(ctx)
        ticker := &Ticker{
                C:         c,
                config:    config,
                writeChan: c,
                cancel:    cancel,
        }
        go ticker.runTicker(ctx)
        return ticker, nil
}

// Run calls the given function repeatedly, using a memoryless.Ticker to wait
// between function calls. It is a convenience function for code that does not
// want to use the channel interface.
func Run(ctx context.Context, f func(), c Config) error {
        ticker, err := MakeTicker(ctx, c)
        if err != nil {
                return err
        }
        defer ticker.Stop()
        for range ticker.C {
                f()
        }
        return nil
}

1	// Package memoryless helps repeated calls to a function be distributed across
2	// time in a memoryless fashion.
3	package memoryless
4
5	// Adapted from https://github.com/m-lab/go/commit/df205a2a463b6624de235da6a61b409567b1ed98
6	// SPDX-License-Identifier: Apache-2.0
7
8	import (
9	"context"
10	"fmt"
11	"math/rand"
12	"time"
13	)
14
15	// Config represents the time we should wait between runs of the function.
16	//
17	// A valid config will have:
18	//
19	// 0 <= Min <= Expected <= Max (or 0 <= Min <= Expected and Max is 0)
20	//
21	// If Max is zero or unset, it will be ignored. If Min is zero or unset, it will
22	// be ignored.
23	type Config struct {
24	// Expected records the expected/mean/average amount of time between runs.
25	Expected time.Duration
26	// Min provides clamping of the randomly produced value. All timers will wait
27	// at least Min time.
28	Min time.Duration
29	// Max provides clamping of the randomly produced value. All timers will take
30	// at most Max time.
31	Max time.Duration
32
33	// Once is provided as a helper, because frequently for unit testing and
34	// integration testing, you only want the "Forever" loop to run once.
35	//
36	// The zero value of this struct has Once set to false, which means the value
37	// only needs to be set explicitly in codepaths where it might be true.
38	Once bool
39	}
40
41	func (c Config) waittime() time.Duration {	1,273✔
42	wt := time.Duration(rand.ExpFloat64() * float64(c.Expected))	1,273✔
43	if wt < c.Min {	2,060✔
44	wt = c.Min	787✔
45	}	787✔
46	if c.Max != 0 && wt > c.Max {	1,758✔
47	wt = c.Max	485✔
48	}	485✔
49	return wt	1,273✔
50	}
51
52	// Check whether the config contrains sensible values. It return an error if the
53	// config makes no mathematical sense, and nil if everything is okay.
54	func (c Config) Check() error {	45✔
55	if !(0 <= c.Min && c.Min <= c.Expected && (c.Max == 0 \|\| c.Expected <= c.Max)) {	85✔
56	return fmt.Errorf(	40✔
57	"memoryless: the arguments to Run make no sense: it should be true that Min <= Expected <= Max (or Min <= Expected and Max is 0), "+	40✔
58	"but that is not true for Min(%v) Expected(%v) Max(%v)",	40✔
59	c.Min, c.Expected, c.Max)	40✔
60	}	40✔
61	return nil	5✔
62	}
63
64	// newTimer constructs and returns a timer. This function assumes that the
65	// config has no errors.
66	func newTimer(c Config) *time.Timer {	1,272✔
67	return time.NewTimer(c.waittime())	1,272✔
68	}	1,272✔
69
70	// NewTimer constructs a single-shot time.Timer that, if repeatedly used to
71	// construct a series of timers, will ensure that the resulting events conform
72	// to the memoryless distribution. For more on how this could and should be
73	// used, see the comments to Ticker. It is intended to be a drop-in replacement
74	// for time.NewTimer.
75	func NewTimer(c Config) (*time.Timer, error) {	9✔
76	if err := c.Check(); err != nil {	17✔
77	return nil, err	8✔
78	}	8✔
79
80	return newTimer(c), nil	1✔
81	}
82
83	// AfterFunc constructs a single-shot time.Timer that, if repeatedly used to
84	// construct a series of timers, will ensure that the resulting events conform
85	// to the memoryless distribution. For more on how this could and should be
86	// used, see the comments to Ticker. It is intended to be a drop-in replacement
87	// for time.AfterFunc.
88	func AfterFunc(c Config, f func()) (*time.Timer, error) {	9✔
89	if err := c.Check(); err != nil {	17✔
90	return nil, err	8✔
91	}	8✔
92
93	return time.AfterFunc(c.waittime(), f), nil	1✔
94	}
95
96	// Ticker is a struct that waits a config.Expected amount of time on average
97	// between sends down the channel C. It has the same interface and requirements
98	// as time.Ticker. Every Ticker created must have its Stop() method called or it
99	// will leak a goroutine.
100	//
101	// The inter-send time is a random variable governed by the exponential
102	// distribution and will generate a memoryless (Poisson) distribution of channel
103	// reads over time, ensuring that a measurement scheme using this ticker has the
104	// PASTA property (Poisson Arrivals See Time Averages). This statistical
105	// guarantee is subject to two caveats:
106	//
107	// Caveat 1 is that, in a nod to the realities of systems needing to have
108	// guarantees, we allow the random wait time to be clamped both above and below.
109	// This means that channel events will be at least config.Min and at most
110	// config.Max apart in time. This clamping causes bias in the timing. For use of
111	// Ticker to be statistically sensible, the clamping should not be too extreme.
112	// The exact mathematical meaning of "too extreme" depends on your situation,
113	// but a nice rule of thumb is config.Min should be at most 10% of expected and
114	// config.Max should be at least 250% of expected. These values mean that less
115	// than 10% of time you will be waiting config.Min and less than 10% of the time
116	// you will be waiting config.Max.
117	//
118	// Caveat 2 is that this assumes that the actions performed between channel
119	// reads take negligible time when compared to the expected wait time.
120	// Memoryless sequences have the property that the times between successive
121	// event starts has the exponential distribution, and the exponential
122	// distribution can generate numbers arbitrarily close to zero (albeit
123	// exponentially infrequently). This code will not send on the channel if the
124	// other end is not ready to receive, which provides another lower bound on
125	// inter-event times. The only other option if the other side of the channel is
126	// not ready to receive would be queueing events in the channel, and that has
127	// some pathological cases we would like to avoid. In particular, queuing can
128	// cause long-term correlations if the queue gets big, which is the exact
129	// opposite of what a memoryless system is trying to achieve.
130	type Ticker struct {
131	C <-chan time.Time // The channel on which the ticks are delivered.
132	config Config
133	writeChan chan<- time.Time
134	cancel func()
135	}
136
137	func (t *Ticker) singleIteration(ctx context.Context) {	1,271✔
138	timer := newTimer(t.config)	1,271✔
139	defer timer.Stop()	1,271✔
140	// Wait until the timer is done or the context is canceled. If both conditions	1,271✔
141	// are true, which case gets called is unspecified.	1,271✔
142	select {	1,271✔
UNCOV 143	case <-ctx.Done():	×
144	// Please don't put code here that assumes that this code path will
145	// definitely execute if the context is done. select {} doesn't promise that
146	// multiple channels will get selected with equal probability, which means
147	// that it could be true that the timer is done AND the context is canceled,
148	// and we have no guarantee that in that case the canceled context case will
149	// be the one that is selected.
150	case <-timer.C:	1,271✔
151	}
152	// Just like time.Ticker, writes to the channel are non-blocking. If a user of
153	// this module can't keep up with the timer they set, that's on them. There
154	// are some potential pathological cases associated with queueing events in
155	// the channel, and we want to avoid them.
156	select {	1,271✔
157	case t.writeChan <- time.Now():	1,004✔
158	default:	267✔
159	}
160	}
161
162	func (t *Ticker) runTicker(ctx context.Context) {	3✔
163	// No matter what, when this function exits the channel should never be written to again.	3✔
164	defer close(t.writeChan)	3✔
165		3✔
166	if t.config.Once {	4✔
167	if ctx.Err() == nil {	2✔
168	t.singleIteration(ctx)	1✔
169	}	1✔
170	return	1✔
171	}
172
173	// When Done() is not closed and the Deadline has not been exceeded, the error
174	// is nil.
175	for ctx.Err() == nil {	1,272✔
176	t.singleIteration(ctx)	1,270✔
177	}	1,270✔
178	}
179
180	// Stop the ticker goroutine.
181	func (t *Ticker) Stop() {	3✔
182	t.cancel()	3✔
183	}	3✔
184
185	// MakeTicker is a deprecated alias for NewTicker
186	var MakeTicker = NewTicker
187
188	// NewTicker creates a new memoryless ticker. The returned struct is compatible
189	// with the time.Ticker struct interface, and everywhere you use a time.Ticker,
190	// you can use a memoryless.Ticker.
191	func NewTicker(ctx context.Context, config Config) (*Ticker, error) {	19✔
192	if err := config.Check(); err != nil {	35✔
193	return nil, err	16✔
194	}	16✔
195	c := make(chan time.Time)	3✔
196	ctx, cancel := context.WithCancel(ctx)	3✔
197	ticker := &Ticker{	3✔
198	C: c,	3✔
199	config: config,	3✔
200	writeChan: c,	3✔
201	cancel: cancel,	3✔
202	}	3✔
203	go ticker.runTicker(ctx)	3✔
204	return ticker, nil	3✔
205	}
206
207	// Run calls the given function repeatedly, using a memoryless.Ticker to wait
208	// between function calls. It is a convenience function for code that does not
209	// want to use the channel interface.
210	func Run(ctx context.Context, f func(), c Config) error {	10✔
211	ticker, err := MakeTicker(ctx, c)	10✔
212	if err != nil {	18✔
213	return err	8✔
214	}	8✔
215	defer ticker.Stop()	2✔
216	for range ticker.C {	1,006✔
217	f()	1,004✔
218	}	1,004✔
219	return nil	2✔
220	}

ooni / probe-cli / 6337355964

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous