Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
39 changed files
with
1,859 additions
and
489 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,156 @@ | ||
| // Copyright © 2020 by PACE Telematics GmbH. All rights reserved. | ||
| // Created at 2020/02/26 by Marius Neugebauer | ||
|
|
||
| package routine_test | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "fmt" | ||
| "io" | ||
| "log" | ||
| "os" | ||
| "os/exec" | ||
| "strings" | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/pace/bricks/pkg/routine" | ||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func Example_clusterBackgroundTask() { | ||
| // This example is an integration test because it requires redis to run. If | ||
| // we are running the short tests just print the output that is expected to | ||
| // circumvent the test runner. Because there is no way to skip an example. | ||
| if testing.Short() { | ||
| fmt.Println("task run 0\ntask run 1\ntask run 2") | ||
| return | ||
| } | ||
|
|
||
| out := make(chan string) | ||
|
|
||
| // start the routine in the background | ||
| cancel := routine.RunNamed(context.Background(), "task", | ||
| func(ctx context.Context) { | ||
| for i := 0; ; i++ { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| default: | ||
| } | ||
| out <- fmt.Sprintf("task run %d", i) | ||
| time.Sleep(100 * time.Millisecond) | ||
| } | ||
| }, | ||
| // KeepRunningOneInstance will cause the routine to be restarted if it | ||
| // finishes. It also will use the default redis database to synchronize | ||
| // with other instances running this routine so that in all instances | ||
| // exactly one routine is running at all time. | ||
| routine.KeepRunningOneInstance(), | ||
| ) | ||
|
|
||
| // Cancel after 3 results. Cancel will only cancel the routine in this | ||
| // instance. It will not cancel the synchronized routines of other | ||
| // instances. | ||
| for i := 0; i < 3; i++ { | ||
| println(<-out) | ||
| } | ||
| cancel() | ||
|
|
||
| // Output: | ||
| // task run 0 | ||
| // task run 1 | ||
| // task run 2 | ||
| } | ||
|
|
||
| func TestIntegrationRunNamed_clusterBackgroundTask(t *testing.T) { | ||
| t.Skip("test not working properly in docker, skipping") | ||
|
|
||
| if testing.Short() { | ||
| t.SkipNow() | ||
| } | ||
|
|
||
| // buffer that allows writing simultaneously | ||
| var buf subprocessOutputBuffer | ||
|
|
||
| // Run 2 processes in the "cluster", that will both try to start the | ||
| // background task in the example function above. The way this task is | ||
| // configured only one process at a time will run the task. But the | ||
| // processes are programmed to exit after 3 iterations of the task. This | ||
| // tests that the second process will take over the execution of the task | ||
| // only after the first process exits. | ||
| var wg sync.WaitGroup | ||
| for i := 0; i < 2; i++ { | ||
| wg.Add(1) | ||
| go func() { | ||
| spawnProcess(&buf) | ||
| wg.Done() | ||
| }() | ||
| } | ||
| wg.Wait() // until both processes are done | ||
|
|
||
| exp := `task run 0 | ||
| task run 1 | ||
| task run 2 | ||
| task run 0 | ||
| task run 1 | ||
| task run 2 | ||
| ` | ||
| assert.Equal(t, exp, buf.String()) | ||
| } | ||
|
|
||
| func spawnProcess(w io.Writer) { | ||
| cmd := exec.Command(os.Args[0], | ||
| "-test.timeout=2s", | ||
| "-test.run=Example_clusterBackgroundTask", | ||
| ) | ||
| cmd.Env = append(os.Environ(), | ||
| "TEST_SUBPROCESS=1", | ||
| "ROUTINE_REDIS_LOCK_TTL=200ms", | ||
| ) | ||
| cmd.Stdout = w | ||
| cmd.Stderr = w | ||
| err := cmd.Run() | ||
| if err != nil { | ||
| _, _ = w.Write([]byte("error starting subprocess: " + err.Error())) | ||
| } | ||
| } | ||
|
|
||
| type subprocessOutputBuffer struct { | ||
| mx sync.Mutex | ||
| buf bytes.Buffer | ||
| } | ||
|
|
||
| func (b *subprocessOutputBuffer) Write(p []byte) (int, error) { | ||
| b.mx.Lock() | ||
| defer b.mx.Unlock() | ||
| // ignore test runner output and some other log lines | ||
| switch s := string(p); { | ||
| case strings.HasPrefix(s, "=== RUN"), | ||
| strings.HasPrefix(s, "--- PASS"), | ||
| strings.HasPrefix(s, "PASS"), | ||
| strings.HasPrefix(s, "coverage: "), | ||
| strings.Contains(s, "Redis connection pool created"): | ||
| return len(p), nil | ||
| } | ||
| return b.buf.Write(p) | ||
| } | ||
|
|
||
| func (b *subprocessOutputBuffer) String() string { | ||
| b.mx.Lock() | ||
| defer b.mx.Unlock() | ||
| return b.buf.String() | ||
| } | ||
|
|
||
| // Prints the string normally so that it can be consumed by the test runner. | ||
| // Additionally go around the test runner in case of a integration test that | ||
| // wants examine the output of another test. | ||
| func println(s string) { | ||
| if os.Getenv("TEST_SUBPROCESS") == "1" { | ||
| // go around the test runner | ||
| _, _ = log.Writer().Write([]byte(s + "\n")) | ||
| } | ||
| fmt.Println(s) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| // Copyright © 2020 by PACE Telematics GmbH. All rights reserved. | ||
| // Created at 2020/02/27 by Marius Neugebauer | ||
|
|
||
| package routine | ||
|
|
||
| import ( | ||
| "time" | ||
|
|
||
| "github.com/caarlos0/env" | ||
| ) | ||
|
|
||
| type config struct { | ||
| RedisLockTTL time.Duration `env:"ROUTINE_REDIS_LOCK_TTL" envDefault:"5s"` | ||
| } | ||
|
|
||
| var cfg config | ||
|
|
||
| func init() { | ||
| if err := env.Parse(&cfg); err != nil { | ||
| panic(err) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,101 @@ | ||
| // Copyright © 2020 by PACE Telematics GmbH. All rights reserved. | ||
| // Created at 2020/02/27 by Marius Neugebauer | ||
|
|
||
| package routine | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/bsm/redislock" | ||
| "github.com/go-redis/redis" | ||
| redisbackend "github.com/pace/bricks/backend/redis" | ||
| ) | ||
|
|
||
| func routineThatKeepsRunningOneInstance(name string, routine func(context.Context)) func(context.Context) { | ||
| return func(ctx context.Context) { | ||
| locker := redislock.New(getDefaultRedisClient()) | ||
| var tryAgainIn time.Duration // zero on first run | ||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| case <-time.After(tryAgainIn): | ||
| } | ||
| lockCtx := obtainLock(ctx, locker, "routine:lock:"+name, cfg.RedisLockTTL) | ||
| if lockCtx != nil { | ||
| routine(lockCtx) | ||
| } | ||
| tryAgainIn = cfg.RedisLockTTL / 5 | ||
| } | ||
| } | ||
| } | ||
|
|
||
| var ( | ||
| initRedisOnce sync.Once | ||
| redisClient *redis.Client | ||
| ) | ||
|
|
||
| func getDefaultRedisClient() *redis.Client { | ||
| initRedisOnce.Do(func() { redisClient = redisbackend.Client() }) | ||
| return redisClient | ||
| } | ||
|
|
||
| // Try to obtain a lock. Return a sub-context of ctx that is canceled once the | ||
| // lock is lost or ctx is done. | ||
| func obtainLock(ctx context.Context, locker *redislock.Client, key string, ttl time.Duration) context.Context { | ||
| // obtain lock | ||
| lock, err := locker.Obtain(key, ttl, nil) | ||
| if err == redislock.ErrNotObtained { | ||
| return nil | ||
| } else if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| // keep up lock, cancel lockCtx otherwise | ||
| lockCtx, cancel := context.WithCancel(ctx) | ||
| go func() { | ||
| keepUpLock(ctx, lock, ttl) | ||
| cancel() | ||
| err := lock.Release() | ||
| if err != nil && err != redislock.ErrLockNotHeld { | ||
| panic(err) | ||
| } | ||
| }() | ||
|
|
||
| return lockCtx | ||
| } | ||
|
|
||
| // Try to keep up a lock for as long as the context is valid. Return once the | ||
| // lock is lost or the context is done. | ||
| func keepUpLock(ctx context.Context, lock *redislock.Lock, refreshTTL time.Duration) { | ||
| refreshInterval := refreshTTL / 5 | ||
| lockRunsOutIn := refreshTTL // initial value must be > refresh interval | ||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
|
|
||
| // Return if the lock runs out and was not refreshed. lockRunsOutIn is | ||
| // always greater than refreshInterval, except the last refresh failed. | ||
| case <-time.After(lockRunsOutIn): | ||
| return | ||
|
|
||
| // Try to refresh lock. | ||
| case <-time.After(refreshInterval): | ||
| } | ||
| if err := lock.Refresh(refreshTTL, nil); err == redislock.ErrNotObtained { | ||
| // Don't return just yet. Get the TTL of the lock and try to | ||
| // refresh for as long as the TTL is not over. | ||
| if lockRunsOutIn, err = lock.TTL(); err != nil { | ||
| panic(err) | ||
| } | ||
| continue | ||
| } else if err != nil { | ||
| panic(err) | ||
| } | ||
| // reset, because the lock was refreshed | ||
| lockRunsOutIn = refreshTTL | ||
| } | ||
| } |
Oops, something went wrong.