12990664691

Committed 27 Jan 2025 02:00PM UTC coverage: 57.056%. First build

Build # 12990664691

Build Type

push

github

Committed by

monstermunchkin

Commit Message

pkg/routine: Allow starting transaction

Run Details

9 of 27 new or added lines in 2 files covered. (33.33%)

5458 of 9566 relevant lines covered (57.06%)

21.65 hits per line

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

0.0

/pkg/routine/instance.go

// Copyright © 2020 by PACE Telematics GmbH. All rights reserved.

package routine

import (
        "context"
        "fmt"
        "time"

        "github.com/getsentry/sentry-go"
        "github.com/pace/bricks/maintenance/errors"
        "github.com/pace/bricks/pkg/lock/redis"

        exponential "github.com/jpillora/backoff"
)

type routineThatKeepsRunningOneInstance struct {
        Name    string
        Routine func(context.Context)

        lockTTL          time.Duration
        retryInterval    time.Duration
        backoff          combinedExponentialBackoff
        num              int64
        startTransaction bool
}

func (r *routineThatKeepsRunningOneInstance) Run(ctx context.Context) {
        // The retry interval is used if we did not get the lock because some
        // other caller got it. The exponential backoff is used if we encounter
        // problems with obtaining the lock, like the Redis not being available.
        // The retry interval is also used if the routine returned regularly, to
        // avoid uncontrollably short restart cycles. If the routine panicked we
        // use exponential backoff as well.
        r.lockTTL = cfg.RedisLockTTL
        r.retryInterval = r.lockTTL / 5
        r.backoff = combinedExponentialBackoff{
                "lock":    &exponential.Backoff{Min: r.retryInterval, Max: 10 * time.Minute},
                "routine": &exponential.Backoff{Min: r.retryInterval, Max: 10 * time.Minute},
        }

        r.num = ctx.Value(ctxNumKey{}).(int64)
        var tryAgainIn time.Duration // zero on first run
        for {
                select {
                case <-ctx.Done():
                        return
                case <-time.After(tryAgainIn):
                }
                // Make sure to cancel the singleRunCtx so that the lock is released
                // after the routine returned.
                singleRunCtx, cancel := context.WithCancel(ctx)
                tryAgainIn = r.singleRun(singleRunCtx)
                cancel()
        }
}

// Performs a single run. That is, to try to obtain the lock and run the routine
// until it returns. Return the backoff duration after which another single run
// should be performed.
func (r *routineThatKeepsRunningOneInstance) singleRun(ctx context.Context) time.Duration {
        l := redis.NewLock("routine:lock:"+r.Name, redis.SetTTL(r.lockTTL))
        lockCtx, cancel, err := l.AcquireAndKeepUp(ctx)
        if err != nil {
                go errors.Handle(ctx, err) // report error to Sentry, non-blocking
                return r.backoff.Duration("lock")
        }
        if lockCtx != nil {
                defer cancel()
                routinePanicked := true
                func() {
                        defer errors.HandleWithCtx(ctx, fmt.Sprintf("routine %d", r.num)) // handle panics

                        var span *sentry.Span

                        if r.startTransaction {
                                span = sentry.StartTransaction(lockCtx, fmt.Sprintf("routine %d", r.num), sentry.WithOpName("function"))
                        } else {
                                span = sentry.StartSpan(lockCtx, "function", sentry.WithDescription(fmt.Sprintf("routine %d", r.num)))
                        }

                        defer span.Finish()

                        r.Routine(span.Context())
                        routinePanicked = false
                }()
                if routinePanicked {
                        return r.backoff.Duration("routine")
                }
        }
        r.backoff.ResetAll()
        return r.retryInterval
}

1	// Copyright © 2020 by PACE Telematics GmbH. All rights reserved.
2
3	package routine
4
5	import (
6	"context"
7	"fmt"
8	"time"
9
10	"github.com/getsentry/sentry-go"
11	"github.com/pace/bricks/maintenance/errors"
12	"github.com/pace/bricks/pkg/lock/redis"
13
14	exponential "github.com/jpillora/backoff"
15	)
16
17	type routineThatKeepsRunningOneInstance struct {
18	Name string
19	Routine func(context.Context)
20
21	lockTTL time.Duration
22	retryInterval time.Duration
23	backoff combinedExponentialBackoff
24	num int64
25	startTransaction bool
26	}
27
28	func (r *routineThatKeepsRunningOneInstance) Run(ctx context.Context) {	×
29	// The retry interval is used if we did not get the lock because some	×
30	// other caller got it. The exponential backoff is used if we encounter	×
31	// problems with obtaining the lock, like the Redis not being available.	×
32	// The retry interval is also used if the routine returned regularly, to	×
33	// avoid uncontrollably short restart cycles. If the routine panicked we	×
34	// use exponential backoff as well.	×
35	r.lockTTL = cfg.RedisLockTTL	×
36	r.retryInterval = r.lockTTL / 5	×
37	r.backoff = combinedExponentialBackoff{	×
38	"lock": &exponential.Backoff{Min: r.retryInterval, Max: 10 * time.Minute},	×
39	"routine": &exponential.Backoff{Min: r.retryInterval, Max: 10 * time.Minute},	×
40	}	×
41		×
42	r.num = ctx.Value(ctxNumKey{}).(int64)	×
43	var tryAgainIn time.Duration // zero on first run	×
44	for {	×
45	select {	×
46	case <-ctx.Done():	×
47	return	×
48	case <-time.After(tryAgainIn):	×
49	}
50	// Make sure to cancel the singleRunCtx so that the lock is released
51	// after the routine returned.
52	singleRunCtx, cancel := context.WithCancel(ctx)	×
53	tryAgainIn = r.singleRun(singleRunCtx)	×
54	cancel()	×
55	}
56	}
57
58	// Performs a single run. That is, to try to obtain the lock and run the routine
59	// until it returns. Return the backoff duration after which another single run
60	// should be performed.
61	func (r *routineThatKeepsRunningOneInstance) singleRun(ctx context.Context) time.Duration {	×
62	l := redis.NewLock("routine:lock:"+r.Name, redis.SetTTL(r.lockTTL))	×
63	lockCtx, cancel, err := l.AcquireAndKeepUp(ctx)	×
64	if err != nil {	×
65	go errors.Handle(ctx, err) // report error to Sentry, non-blocking	×
66	return r.backoff.Duration("lock")	×
67	}	×
68	if lockCtx != nil {	×
69	defer cancel()	×
70	routinePanicked := true	×
71	func() {	×
72	defer errors.HandleWithCtx(ctx, fmt.Sprintf("routine %d", r.num)) // handle panics	×
73		×
NEW 74	var span *sentry.Span	×
NEW 75		×
NEW 76	if r.startTransaction {	×
NEW 77	span = sentry.StartTransaction(lockCtx, fmt.Sprintf("routine %d", r.num), sentry.WithOpName("function"))	×
NEW 78	} else {	×
NEW 79	span = sentry.StartSpan(lockCtx, "function", sentry.WithDescription(fmt.Sprintf("routine %d", r.num)))	×
NEW 80	}	×
81
82	defer span.Finish()	×
83		×
84	r.Routine(span.Context())	×
85	routinePanicked = false	×
86	}()
87	if routinePanicked {	×
88	return r.backoff.Duration("routine")	×
89	}	×
90	}
91	r.backoff.ResetAll()	×
92	return r.retryInterval	×
93	}

pace / bricks / 12990664691

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