forked from kevinburke/rickover
/
main.go
92 lines (79 loc) · 2.44 KB
/
main.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
// Command dequeuer dequeues jobs and sends them to a downstream server.
package main
import (
"context"
"fmt"
"log"
"os"
"os/signal"
"time"
"github.com/kevinburke/go-simple-metrics"
"github.com/kevinburke/rickover/config"
"github.com/kevinburke/rickover/dequeuer"
"github.com/kevinburke/rickover/models/db"
"github.com/kevinburke/rickover/services"
"github.com/kevinburke/rickover/setup"
"golang.org/x/sync/errgroup"
"golang.org/x/sys/unix"
)
func checkError(err error) {
if err != nil {
log.Fatal(err)
}
}
func main() {
dbConns, err := config.GetInt("PG_WORKER_POOL_SIZE")
if err != nil {
log.Printf("Error getting database pool size: %s. Defaulting to 20", err)
dbConns = 20
}
err = setup.DB(db.DefaultConnection, dbConns)
checkError(err)
go setup.MeasureActiveQueries(1 * time.Second)
go setup.MeasureQueueDepth(5 * time.Second)
go setup.MeasureInProgressJobs(1 * time.Second)
// Every minute, check for in-progress jobs that haven't been updated for
// 7 minutes, and mark them as failed.
go services.WatchStuckJobs(1*time.Minute, 7*time.Minute)
// We're going to make a lot of requests to the same downstream service.
httpConns, err := config.GetInt("HTTP_MAX_IDLE_CONNS")
if err == nil {
config.SetMaxIdleConnsPerHost(httpConns)
} else {
config.SetMaxIdleConnsPerHost(100)
}
metrics.Namespace = "rickover.dequeuer"
metrics.Start("worker", os.Getenv("LIBRATO_EMAIL_ACCOUNT"))
downstreamPassword := os.Getenv("DOWNSTREAM_WORKER_AUTH")
if downstreamPassword == "" {
log.Printf("No DOWNSTREAM_WORKER_AUTH configured, setting an empty password for auth")
}
parsedUrl := config.GetURLOrBail("DOWNSTREAM_URL")
jp := services.NewJobProcessor(parsedUrl.String(), downstreamPassword)
// This creates a pool of dequeuers and starts them.
pools, err := dequeuer.CreatePools(jp, 200*time.Millisecond)
checkError(err)
sigterm := make(chan os.Signal, 1)
signal.Notify(sigterm, unix.SIGINT, unix.SIGTERM)
sig := <-sigterm
fmt.Printf("Caught signal %v, shutting down...\n", sig)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
g, errctx := errgroup.WithContext(ctx)
for _, p := range pools {
if p != nil {
p := p
g.Go(func() error {
err := p.Shutdown(errctx)
if err != nil {
log.Printf("Error shutting down pool: %v", err.Error())
}
return err
})
}
}
if err := g.Wait(); err != nil {
log.Fatal(err)
}
log.Println("All pools shut down. Quitting.")
}