-
Notifications
You must be signed in to change notification settings - Fork 1
/
serv.go
132 lines (119 loc) · 4.09 KB
/
serv.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
package servicegroupservice
import (
"context"
"errors"
"fmt"
"strings"
"github.com/ScoreTrak/ScoreTrak/pkg/config"
"github.com/ScoreTrak/ScoreTrak/pkg/platform"
"github.com/ScoreTrak/ScoreTrak/pkg/platform/worker"
"github.com/ScoreTrak/ScoreTrak/pkg/queue"
"github.com/ScoreTrak/ScoreTrak/pkg/servicegroup"
repo2 "github.com/ScoreTrak/ScoreTrak/pkg/servicegroup/servicegrouprepo"
"github.com/gofrs/uuid"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
type Serv interface {
Delete(ctx context.Context, id uuid.UUID) error
GetAll(ctx context.Context) ([]*servicegroup.ServiceGroup, error)
GetByID(ctx context.Context, id uuid.UUID) (*servicegroup.ServiceGroup, error)
Store(ctx context.Context, u *servicegroup.ServiceGroup) error
Update(ctx context.Context, u *servicegroup.ServiceGroup) error
Redeploy(ctx context.Context, id uuid.UUID) error
}
type serviceGroupServ struct {
repo repo2.Repo
p platform.Platform
q queue.WorkerQueue
}
var ErrRedeployNotAllowed = errors.New("redeploy is not allowed when platform or queue are not specified")
var ErrRedeployDisableGroup = errors.New("check_service group must first be disabled")
func (svc *serviceGroupServ) Redeploy(ctx context.Context, id uuid.UUID) error {
if svc.p == nil || config.GetStaticConfig().Queue.Use == queue.None {
return ErrRedeployNotAllowed
}
serGrp, err := svc.GetByID(ctx, id)
if err != nil {
return err
}
if *serGrp.Enabled {
return ErrRedeployDisableGroup
}
wr := worker.Info{Topic: serGrp.Name, Label: serGrp.Label}
err = svc.p.RemoveWorkers(ctx, wr)
if err != nil && !strings.Contains(err.Error(), "not found") {
return fmt.Errorf("scoretrak encountered an error while removing the workersvc. Please, delete the workers manually. Details: %w", err)
}
err = svc.p.DeployWorkers(ctx, wr)
if err != nil {
return fmt.Errorf("scoretrak encountered an error while deploying the workersvc. Please, investigate the issue, or create the workers manually. Details: %w", err)
}
return nil
}
func NewServiceGroupServ(repo repo2.Repo, plat platform.Platform, q queue.WorkerQueue) Serv {
return &serviceGroupServ{
repo: repo, p: plat, q: q,
}
}
func (svc *serviceGroupServ) Delete(ctx context.Context, id uuid.UUID) error {
serviceGrp, err := svc.GetByID(ctx, id)
if err != nil {
return status.Errorf(
codes.NotFound,
fmt.Sprintf("Service Group not found: %v", err),
)
}
err = svc.repo.Delete(ctx, id)
if err != nil {
return err
}
if svc.p != nil && config.GetStaticConfig().Queue.Use != queue.None {
wr := worker.Info{Topic: serviceGrp.Name, Label: serviceGrp.Label}
err := svc.p.RemoveWorkers(ctx, wr)
if err != nil {
return fmt.Errorf("element was removed from database, however, we were unable to remove workers from platform. Error: %w", err)
}
}
return nil
}
func (svc *serviceGroupServ) GetAll(ctx context.Context) ([]*servicegroup.ServiceGroup, error) {
return svc.repo.GetAll(ctx)
}
func (svc *serviceGroupServ) GetByID(ctx context.Context, id uuid.UUID) (*servicegroup.ServiceGroup, error) {
return svc.repo.GetByID(ctx, id)
}
func (svc *serviceGroupServ) Store(ctx context.Context, u *servicegroup.ServiceGroup) error {
if svc.p != nil && !u.SkipHelper && config.GetStaticConfig().Queue.Use != queue.None {
if u.Enabled != nil && *u.Enabled {
return status.Errorf(
codes.FailedPrecondition,
"if you are letting scoretrak manage the workers, 'Enabled' can be set to true, only after workers are deployed.",
)
}
wr := worker.Info{Topic: u.Name, Label: u.Label}
err := svc.p.DeployWorkers(ctx, wr)
if err != nil {
return err
}
}
return svc.repo.Store(ctx, u)
}
func (svc *serviceGroupServ) Update(ctx context.Context, u *servicegroup.ServiceGroup) error {
serviceGrp, err := svc.GetByID(ctx, u.ID)
if err != nil {
return status.Errorf(
codes.NotFound,
fmt.Sprintf("Service Group not found: %v", err),
)
}
if !u.SkipHelper && config.GetStaticConfig().Queue.Use != queue.None {
if u.Enabled != nil && *u.Enabled && !*serviceGrp.Enabled {
err = svc.q.Ping(serviceGrp)
if err != nil {
return err
}
}
}
return svc.repo.Update(ctx, u)
}