forked from Shyp/rickover
-
Notifications
You must be signed in to change notification settings - Fork 2
/
main.go
77 lines (69 loc) · 2.09 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
// Command dequeuer dequeues jobs and sends them to a downstream server.
package main
import (
"context"
"flag"
"fmt"
"log"
"os"
"os/signal"
"github.com/kevinburke/handlers"
"github.com/kevinburke/rickover/config"
"github.com/kevinburke/rickover/dequeuer"
"github.com/kevinburke/rickover/metrics"
"github.com/kevinburke/rickover/services"
"golang.org/x/sys/unix"
)
func checkError(err error) {
if err != nil {
log.Fatal(err)
}
}
func main() {
flag.Parse()
logger := handlers.Logger
ctx, cancel := context.WithCancel(context.Background())
go func() {
sigterm := make(chan os.Signal, 1)
signal.Notify(sigterm, unix.SIGINT, unix.SIGTERM)
sig := <-sigterm
fmt.Printf("Caught signal %v, shutting down...\n", sig)
cancel()
}()
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
}
// 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)
}
go metrics.Run(ctx, metrics.LibratoConfig{
Namespace: "rickover.dequeuer",
Source: "worker",
Email: os.Getenv("LIBRATO_EMAIL_ACCOUNT"),
})
parsedUrl := config.GetURLOrBail("DOWNSTREAM_URL")
downstreamPassword := os.Getenv("DOWNSTREAM_WORKER_AUTH")
if downstreamPassword == "" {
logger.Warn("No DOWNSTREAM_WORKER_AUTH configured, setting an empty password for auth")
}
handler := services.NewDownstreamHandler(logger, parsedUrl.String(), downstreamPassword)
srv, err := dequeuer.New(ctx, dequeuer.Config{
Logger: logger,
NumConns: dbConns,
Processor: services.NewJobProcessor(handler),
StuckJobTimeout: dequeuer.DefaultStuckJobTimeout,
DisableMetaShutdown: os.Getenv("DISABLE_META_SHUTDOWN") == "true",
})
checkError(err)
if err := srv.Run(ctx); err != nil && err != context.Canceled {
logger.Error("error running dequeuer", "err", err)
os.Exit(1)
}
logger.Info("All pools shut down. Quitting.")
}