Skip to content

Commit 1a6173f

Browse files
sguiheuxsguiheux
authored
feat(cdn): use logrus to send worker and service log (#5153)
1 parent 4285535 commit 1a6173f

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

45 files changed

+1374
-111
lines changed

‎engine/api/api.go

+9
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ import (
4949
"github.com/ovh/cds/engine/api/worker"
5050
"github.com/ovh/cds/engine/api/workermodel"
5151
"github.com/ovh/cds/engine/api/workflow"
52+
"github.com/ovh/cds/engine/cdn"
5253
"github.com/ovh/cds/engine/service"
5354
"github.com/ovh/cds/sdk"
5455
"github.com/ovh/cds/sdk/cdsclient"
@@ -192,6 +193,7 @@ type Configuration struct {
192193
StepMaxSize int64 `toml:"stepMaxSize" default:"15728640" comment:"Max step logs size in bytes (default: 15MB)" json:"stepMaxSize"`
193194
ServiceMaxSize int64 `toml:"serviceMaxSize" default:"15728640" comment:"Max service logs size in bytes (default: 15MB)" json:"serviceMaxSize"`
194195
} `toml:"log" json:"log" comment:"###########################\n Log settings.\n##########################"`
196+
CDN cdn.Configuration `toml:"cdn" json:"cdn" comment:"###########################\n CDN settings.\n##########################"`
195197
}
196198

197199
// ServiceConfiguration is the configuration of external service
@@ -872,6 +874,13 @@ func (a *API) Serve(ctx context.Context) error {
872874
log.Error(ctx, "api> heap dump uploaded to %s", s)
873875
}()
874876

877+
cdsService := &cdn.Service{
878+
Cfg: a.Config.CDN,
879+
Db: a.mustDB(),
880+
Cache: a.Cache,
881+
}
882+
cdsService.RunTcpLogServer(ctx)
883+
875884
log.Info(ctx, "Starting CDS API HTTP Server on %s:%d", a.Config.HTTP.Addr, a.Config.HTTP.Port)
876885
if err := s.ListenAndServe(); err != nil {
877886
return fmt.Errorf("Cannot start HTTP server: %v", err)

‎engine/api/database/gorpmapping/encryption.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -108,8 +108,8 @@ func updateEncryptedData(db gorp.SqlExecutor, i interface{}) error {
108108
updateSlice = append(updateSlice, encryptedColumns[f]+" = $"+strconv.Itoa(c))
109109
c++
110110
}
111-
112-
query := fmt.Sprintf("UPDATE %s SET %s WHERE %s = %v", table, strings.Join(updateSlice, ","), key, id)
111+
encryptedContentArgs = append(encryptedContentArgs, id)
112+
query := fmt.Sprintf("UPDATE %s SET %s WHERE %s = $%d", table, strings.Join(updateSlice, ","), key, c)
113113
res, err := db.Exec(query, encryptedContentArgs...)
114114
if err != nil {
115115
return sdk.WithStack(err)

‎engine/api/services.go

+1
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,7 @@ func (api *API) postServiceRegisterHandler() service.Handler {
9494
}
9595

9696
srv.Uptodate = data.Version == sdk.VERSION
97+
srv.LogServer = api.Config.CDN.TCP
9798

9899
return service.WriteJSON(w, srv, http.StatusOK)
99100
}

‎engine/api/services/const.go

+1
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ const (
88
TypeVCS = "vcs"
99
TypeAPI = "api"
1010
TypeUI = "ui"
11+
TypeCDN = "cdn"
1112
TypeHatchery = "hatchery"
1213
TypeDBMigrate = "dbmigrate"
1314
)

‎engine/api/worker.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -213,7 +213,7 @@ func (api *API) workerWaitingHandler() service.Handler {
213213
return nil
214214
}
215215

216-
if err := worker.SetStatus(api.mustDB(), wk.ID, sdk.StatusWaiting); err != nil {
216+
if err := worker.SetStatus(ctx, api.mustDB(), wk.ID, sdk.StatusWaiting); err != nil {
217217
return sdk.WrapError(err, "cannot update worker %s", wk.ID)
218218
}
219219
return nil
@@ -254,7 +254,7 @@ func DisableWorker(ctx context.Context, db *gorp.DbMap, id string) error {
254254
log.Info(ctx, "DisableWorker> Worker %s crashed while building %d !", name, jobID.Int64)
255255
}
256256

257-
if err := worker.SetStatus(tx, id, sdk.StatusDisabled); err != nil {
257+
if err := worker.SetStatus(ctx, tx, id, sdk.StatusDisabled); err != nil {
258258
cause := sdk.Cause(err)
259259
if cause == worker.ErrNoWorker || cause == sql.ErrNoRows {
260260
return sdk.WrapError(sdk.ErrWrongRequest, "DisableWorker> worker %s does not exists", id)

‎engine/api/worker/dao.go

+127-25
Original file line numberDiff line numberDiff line change
@@ -11,8 +11,13 @@ import (
1111
"github.com/ovh/cds/sdk"
1212
)
1313

14-
func Insert(db gorp.SqlExecutor, w *sdk.Worker) error {
15-
return gorpmapping.Insert(db, w)
14+
func Insert(ctx context.Context, db gorp.SqlExecutor, w *sdk.Worker) error {
15+
dbData := &dbWorker{Worker: *w}
16+
if err := gorpmapping.InsertAndSign(ctx, db, dbData); err != nil {
17+
return err
18+
}
19+
*w = dbData.Worker
20+
return nil
1621
}
1722

1823
// Delete remove worker from database, it also removes the associated access_token
@@ -37,83 +42,180 @@ func Delete(db gorp.SqlExecutor, id string) error {
3742

3843
func LoadByConsumerID(ctx context.Context, db gorp.SqlExecutor, id string) (*sdk.Worker, error) {
3944
query := gorpmapping.NewQuery("SELECT * FROM worker WHERE auth_consumer_id = $1").Args(id)
40-
var w sdk.Worker
45+
var w dbWorker
4146
found, err := gorpmapping.Get(ctx, db, query, &w)
4247
if err != nil {
4348
return nil, err
4449
}
4550
if !found {
4651
return nil, sdk.WithStack(sdk.ErrNotFound)
4752
}
48-
return &w, nil
53+
isValid, err := gorpmapping.CheckSignature(w, w.Signature)
54+
if err != nil {
55+
return nil, err
56+
}
57+
if !isValid {
58+
return nil, sdk.WithStack(sdk.ErrInvalidData)
59+
}
60+
return &w.Worker, nil
4961
}
5062

5163
func LoadByID(ctx context.Context, db gorp.SqlExecutor, id string) (*sdk.Worker, error) {
5264
query := gorpmapping.NewQuery("SELECT * FROM worker WHERE id = $1").Args(id)
53-
var w sdk.Worker
65+
var w dbWorker
5466
found, err := gorpmapping.Get(ctx, db, query, &w)
5567
if err != nil {
5668
return nil, err
5769
}
5870
if !found {
5971
return nil, sdk.WithStack(sdk.ErrNotFound)
6072
}
61-
return &w, nil
73+
isValid, err := gorpmapping.CheckSignature(w, w.Signature)
74+
if err != nil {
75+
return nil, err
76+
}
77+
if !isValid {
78+
return nil, sdk.WithStack(sdk.ErrInvalidData)
79+
}
80+
return &w.Worker, nil
6281
}
6382

6483
func LoadAll(ctx context.Context, db gorp.SqlExecutor) ([]sdk.Worker, error) {
65-
var workers []sdk.Worker
84+
var wks []dbWorker
6685
query := gorpmapping.NewQuery(`SELECT * FROM worker ORDER BY name ASC`)
67-
if err := gorpmapping.GetAll(ctx, db, query, &workers); err != nil {
86+
if err := gorpmapping.GetAll(ctx, db, query, &wks); err != nil {
6887
return nil, err
6988
}
89+
workers := make([]sdk.Worker, len(wks))
90+
for i := range wks {
91+
isValid, err := gorpmapping.CheckSignature(wks[i], wks[i].Signature)
92+
if err != nil {
93+
return nil, err
94+
}
95+
if !isValid {
96+
return nil, sdk.WithStack(sdk.ErrInvalidData)
97+
}
98+
workers[i] = wks[i].Worker
99+
}
70100
return workers, nil
71101
}
72102

73103
func LoadByHatcheryID(ctx context.Context, db gorp.SqlExecutor, hatcheryID int64) ([]sdk.Worker, error) {
74-
var workers []sdk.Worker
104+
var wks []dbWorker
75105
query := gorpmapping.NewQuery(`SELECT * FROM worker WHERE hatchery_id = $1 ORDER BY name ASC`).Args(hatcheryID)
76-
if err := gorpmapping.GetAll(ctx, db, query, &workers); err != nil {
106+
if err := gorpmapping.GetAll(ctx, db, query, &wks); err != nil {
77107
return nil, err
78108
}
109+
workers := make([]sdk.Worker, len(wks))
110+
for i := range wks {
111+
isValid, err := gorpmapping.CheckSignature(wks[i], wks[i].Signature)
112+
if err != nil {
113+
return nil, err
114+
}
115+
if !isValid {
116+
return nil, sdk.WithStack(sdk.ErrInvalidData)
117+
}
118+
workers[i] = wks[i].Worker
119+
}
79120
return workers, nil
80121
}
81122

82123
func LoadDeadWorkers(ctx context.Context, db gorp.SqlExecutor, timeout float64, status []string) ([]sdk.Worker, error) {
83-
var workers []sdk.Worker
124+
var wks []dbWorker
84125
query := gorpmapping.NewQuery(`SELECT *
85126
FROM worker
86127
WHERE status = ANY(string_to_array($1, ',')::text[])
87128
AND now() - last_beat > $2 * INTERVAL '1' SECOND
88129
ORDER BY last_beat ASC`).Args(strings.Join(status, ","), timeout)
89-
if err := gorpmapping.GetAll(ctx, db, query, &workers); err != nil {
130+
if err := gorpmapping.GetAll(ctx, db, query, &wks); err != nil {
90131
return nil, err
91132
}
133+
workers := make([]sdk.Worker, len(wks))
134+
for i := range wks {
135+
isValid, err := gorpmapping.CheckSignature(wks[i], wks[i].Signature)
136+
if err != nil {
137+
return nil, err
138+
}
139+
if !isValid {
140+
return nil, sdk.WithStack(sdk.ErrInvalidData)
141+
}
142+
workers[i] = wks[i].Worker
143+
}
92144
return workers, nil
93145
}
94146

95147
// SetStatus sets job_run_id and status to building on given worker
96-
func SetStatus(db gorp.SqlExecutor, workerID string, status string) error {
97-
query := `UPDATE worker SET status = $1 WHERE id = $2`
148+
func SetStatus(ctx context.Context, db gorp.SqlExecutor, workerID string, status string) error {
149+
w, err := LoadByID(ctx, db, workerID)
150+
if err != nil {
151+
return err
152+
}
153+
w.Status = status
98154
if status == sdk.StatusBuilding || status == sdk.StatusWaiting {
99-
query = `UPDATE worker SET status = $1, job_run_id = NULL WHERE id = $2`
155+
w.JobRunID = nil
100156
}
101-
102-
if _, err := db.Exec(query, status, workerID); err != nil {
103-
return sdk.WithStack(err)
157+
dbData := &dbWorker{Worker: *w}
158+
if err := gorpmapping.UpdateAndSign(ctx, db, dbData); err != nil {
159+
return err
104160
}
105161
return nil
106162
}
107163

108164
// SetToBuilding sets job_run_id and status to building on given worker
109-
func SetToBuilding(db gorp.SqlExecutor, workerID string, jobRunID int64) error {
110-
query := `UPDATE worker SET status = $1, job_run_id = $2 WHERE id = $3`
165+
func SetToBuilding(ctx context.Context, db gorp.SqlExecutor, workerID string, jobRunID int64, key []byte) error {
166+
w, err := LoadByID(ctx, db, workerID)
167+
if err != nil {
168+
return err
169+
}
170+
w.Status = sdk.StatusBuilding
171+
w.JobRunID = &jobRunID
172+
w.PrivateKey = key
111173

112-
res, errE := db.Exec(query, sdk.StatusBuilding, jobRunID, workerID)
113-
if errE != nil {
114-
return sdk.WithStack(errE)
174+
dbData := &dbWorker{Worker: *w}
175+
if err := gorpmapping.UpdateAndSign(ctx, db, dbData); err != nil {
176+
return err
115177
}
178+
return nil
179+
}
180+
181+
// LoadWorkerByIDWithDecryptKey load worker with decrypted private key
182+
func LoadWorkerByIDWithDecryptKey(ctx context.Context, db gorp.SqlExecutor, workerID string) (*sdk.Worker, error) {
183+
var work dbWorker
184+
query := gorpmapping.NewQuery(`SELECT * FROM worker WHERE id = $1`).Args(workerID)
185+
found, err := gorpmapping.Get(ctx, db, query, &work, gorpmapping.GetOptions.WithDecryption)
186+
if err != nil {
187+
return nil, err
188+
}
189+
if !found {
190+
return nil, sdk.WithStack(sdk.ErrNotFound)
191+
}
192+
isValid, err := gorpmapping.CheckSignature(work, work.Signature)
193+
if err != nil {
194+
return nil, err
195+
}
196+
if !isValid {
197+
return nil, sdk.WithStack(sdk.ErrInvalidData)
198+
}
199+
return &work.Worker, err
200+
}
116201

117-
_, err := res.RowsAffected()
118-
return err
202+
// LoadWorkerByName load worker by name
203+
func LoadWorkerByName(ctx context.Context, db gorp.SqlExecutor, workerName string) (*sdk.Worker, error) {
204+
var work dbWorker
205+
query := gorpmapping.NewQuery(`SELECT * FROM worker WHERE name = $1`).Args(workerName)
206+
found, err := gorpmapping.Get(ctx, db, query, &work)
207+
if err != nil {
208+
return nil, err
209+
}
210+
if !found {
211+
return nil, sdk.WithStack(sdk.ErrNotFound)
212+
}
213+
isValid, err := gorpmapping.CheckSignature(work, work.Signature)
214+
if err != nil {
215+
return nil, err
216+
}
217+
if !isValid {
218+
return nil, sdk.WithStack(sdk.ErrInvalidData)
219+
}
220+
return &work.Worker, err
119221
}

‎engine/api/worker/gorp_model.go

+22
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
package worker
2+
3+
import (
4+
"github.com/ovh/cds/engine/api/database/gorpmapping"
5+
"github.com/ovh/cds/sdk"
6+
)
7+
8+
type dbWorker struct {
9+
gorpmapping.SignedEntity
10+
sdk.Worker
11+
}
12+
13+
func init() {
14+
gorpmapping.Register(gorpmapping.New(dbWorker{}, "worker", false, "id"))
15+
}
16+
17+
func (e dbWorker) Canonical() gorpmapping.CanonicalForms {
18+
var _ = []interface{}{e.ID, e.Name}
19+
return gorpmapping.CanonicalForms{
20+
"{{print .ID}}{{.Name}}",
21+
}
22+
}

‎engine/api/worker/heartbeat.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ func DisableDeadWorkers(ctx context.Context, db *gorp.DbMap) error {
1919
}
2020
for i := range workers {
2121
log.Debug("Disable worker %s[%s] LastBeat:%v status:%s", workers[i].Name, workers[i].ID, workers[i].LastBeat, workers[i].Status)
22-
if errD := SetStatus(db, workers[i].ID, sdk.StatusDisabled); errD != nil {
22+
if errD := SetStatus(ctx, db, workers[i].ID, sdk.StatusDisabled); errD != nil {
2323
log.Warning(ctx, "Cannot disable worker %v: %v", workers[i].ID, errD)
2424
}
2525
}

‎engine/api/worker/init.go

-6
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,6 @@ import (
77
"github.com/go-gorp/gorp"
88

99
"github.com/ovh/cds/engine/api/cache"
10-
"github.com/ovh/cds/engine/api/database/gorpmapping"
11-
"github.com/ovh/cds/sdk"
1210
"github.com/ovh/cds/sdk/log"
1311
)
1412

@@ -39,7 +37,3 @@ func Initialize(c context.Context, DBFunc func() *gorp.DbMap, store cache.Store)
3937
}
4038
}
4139
}
42-
43-
func init() {
44-
gorpmapping.Register(gorpmapping.New(sdk.Worker{}, "worker", false, "id"))
45-
}

‎engine/api/worker/registration.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -106,7 +106,7 @@ func RegisterWorker(ctx context.Context, db gorp.SqlExecutor, store cache.Store,
106106

107107
w.Uptodate = registrationForm.Version == sdk.VERSION
108108

109-
if err := Insert(db, w); err != nil {
109+
if err := Insert(ctx, db, w); err != nil {
110110
return nil, err
111111
}
112112

‎engine/api/worker/worker_test.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ func TestDAO(t *testing.T) {
3838
Status: sdk.StatusWaiting,
3939
}
4040

41-
if err := worker.Insert(db, w); err != nil {
41+
if err := worker.Insert(context.TODO(), db, w); err != nil {
4242
t.Fatalf("Cannot insert worker %+v: %v", w, err)
4343
}
4444

@@ -57,7 +57,7 @@ func TestDAO(t *testing.T) {
5757
assert.Equal(t, "foofoo", wk.ID)
5858
}
5959

60-
test.NoError(t, worker.SetStatus(db, wk.ID, sdk.StatusBuilding))
60+
test.NoError(t, worker.SetStatus(context.TODO(), db, wk.ID, sdk.StatusBuilding))
6161
test.NoError(t, worker.RefreshWorker(db, wk.ID))
6262
}
6363

0 commit comments

Comments
 (0)