-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.go
145 lines (112 loc) · 3.86 KB
/
app.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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
package app
import (
"net/http"
"net/url"
"os"
"os/signal"
"syscall"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/credentials"
"github.com/suchimauz/walg-k8s-cron-backup/internal/config"
"github.com/suchimauz/walg-k8s-cron-backup/pkg/kube"
"github.com/suchimauz/walg-k8s-cron-backup/pkg/storage"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api/v5"
cr "github.com/robfig/cron/v3"
cjobs "github.com/suchimauz/walg-k8s-cron-backup/internal/job"
klog "github.com/suchimauz/walg-k8s-cron-backup/pkg/logger"
)
func Run(dotenv func()) {
// Initialize config
cfg, err := config.Init(dotenv)
if err != nil {
klog.Errorf("[ENV] %s", err.Error())
return
}
// Initialize Kubernetes tls config for set insecure: true of false from config
tlsClientConfig := &rest.TLSClientConfig{Insecure: cfg.Kubernetes.Insecure}
// Initialize Kubernetes BearerToken config
kubeConfig := &rest.Config{
Host: cfg.Kubernetes.Host,
APIPath: cfg.Kubernetes.ApiVersion,
BearerToken: cfg.Kubernetes.BearerToken,
TLSClientConfig: *tlsClientConfig,
}
// Create new kubernetes client from kubeConfig
clientset, err := kubernetes.NewForConfig(kubeConfig)
if err != nil {
klog.Errorf("[KubeConfig] %s", err.Error())
return
}
// Create new local KubeJob pkg object
kjob, err := kube.NewKubeJob(clientset, kubeConfig, cfg.Kubernetes.Namespace,
cfg.Kubernetes.LabelSelector, cfg.Kubernetes.ContainerName)
if err != nil {
klog.Errorf("[KubeJob] %s", err.Error())
return
}
// Create new object for telegram api when one of notification is enabled
var tgbot *tgbotapi.BotAPI
// Create new http client for telegram api
tgclient := &http.Client{}
// When http proxy for telegram api is declared,
if cfg.Telegram.HttpProxy != "" {
proxy, err := url.Parse(cfg.Telegram.HttpProxy)
if err != nil {
klog.Errorf("[TelegramBotApi] Proxy: %s", err.Error())
}
transport := &http.Transport{}
transport.Proxy = http.ProxyURL(proxy)
tgclient.Transport = transport
}
if cfg.Telegram.NotificationsEnabled() {
tgbot, err = tgbotapi.NewBotAPIWithClient(cfg.Telegram.BotToken, cfg.Telegram.ApiEndpoint, tgclient)
if err != nil {
klog.Errorf("[TelegramBotApi] %s", err.Error())
return
}
}
// Init storage provider - minio if save logs is enabled
var storageProvider storage.Provider
if cfg.FileStorageRequired() {
storageProvider, err = newStorageProvider(cfg)
if err != nil {
klog.Errorf("[FileStorage] Provider: %s", err.Error())
return
}
}
// Make new cron object, calls constructor
cron := cr.New(cr.WithSeconds(), cr.WithLocation(config.TimeZone))
// Insert jobs to cron
jobIds, err := cjobs.InsertJobs(cron, cfg, kjob, tgbot, storageProvider)
if err != nil {
klog.Errorf("[Cron] Error inserting jobs: %s", err.Error())
return
}
// Start the cron scheduler in its own goroutine
cron.Start()
klog.Infof("[Cron] Started! JobIds %v", jobIds)
// Graceful Shutdown
// Make new channel of size = 1
quit := make(chan os.Signal, 1)
// Listen system 15 and 2 signals, when one of they called, send info to quit channel
signal.Notify(quit, syscall.SIGTERM, syscall.SIGINT)
// Read channel, this block of code lock this thread, until someone writes to the channel
<-quit
// When someone call SIGTERM or SIGINT signals, we'll get to here
// cron.Stop() -> Wait jobs and stop
cron.Stop()
klog.Info("[Cron] Stopped! Exit")
}
func newStorageProvider(cfg *config.Config) (storage.Provider, error) {
client, err := minio.New(cfg.FileStorage.Endpoint, &minio.Options{
Creds: credentials.NewStaticV4(cfg.FileStorage.AccessKey, cfg.FileStorage.SecretKey, ""),
Secure: cfg.FileStorage.Secure,
})
if err != nil {
return nil, err
}
provider := storage.NewFileStorage(client, cfg.FileStorage.Bucket, cfg.FileStorage.Endpoint)
return provider, nil
}