-
Notifications
You must be signed in to change notification settings - Fork 0
/
persistence.go
59 lines (53 loc) · 1.68 KB
/
persistence.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
package fixtures
import (
"k8s.io/client-go/kubernetes"
"upper.io/db.v3/lib/sqlbuilder"
"github.com/argoproj/argo/config"
"github.com/argoproj/argo/persist/sqldb"
"github.com/argoproj/argo/util/instanceid"
)
type Persistence struct {
session sqlbuilder.Database
offloadNodeStatusRepo sqldb.OffloadNodeStatusRepo
workflowArchive sqldb.WorkflowArchive
}
func newPersistence(kubeClient kubernetes.Interface) *Persistence {
configController := config.NewController(Namespace, "workflow-controller-configmap", kubeClient)
wcConfig, err := configController.Get()
if err != nil {
panic(err)
}
persistence := wcConfig.Persistence
if persistence != nil {
if persistence.PostgreSQL != nil {
persistence.PostgreSQL.Host = "localhost"
}
if persistence.MySQL != nil {
persistence.MySQL.Host = "localhost"
}
session, tableName, err := sqldb.CreateDBSession(kubeClient, Namespace, persistence)
if err != nil {
panic(err)
}
offloadNodeStatusRepo, err := sqldb.NewOffloadNodeStatusRepo(session, persistence.GetClusterName(), tableName)
if err != nil {
panic(err)
}
instanceIDService := instanceid.NewService(wcConfig.InstanceID)
workflowArchive := sqldb.NewWorkflowArchive(session, persistence.GetClusterName(), Namespace, instanceIDService)
return &Persistence{session, offloadNodeStatusRepo, workflowArchive}
} else {
return &Persistence{offloadNodeStatusRepo: sqldb.ExplosiveOffloadNodeStatusRepo, workflowArchive: sqldb.NullWorkflowArchive}
}
}
func (s *Persistence) IsEnabled() bool {
return s.offloadNodeStatusRepo.IsEnabled()
}
func (s *Persistence) Close() {
if s.IsEnabled() {
err := s.session.Close()
if err != nil {
panic(err)
}
}
}