/
handlerService.go
381 lines (327 loc) · 12.1 KB
/
handlerService.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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
// Copyright (c) 2016 OpenM++
// This code is licensed under the MIT license (see LICENSE.txt for details)
package main
import (
"net/http"
"strconv"
"github.com/openmpp/go/ompp/db"
"github.com/openmpp/go/ompp/helper"
"github.com/openmpp/go/ompp/omppLog"
)
// serviceConfigHandler return service configuration, including configuration of model catalog and run catalog.
// GET /api/service/config
func serviceConfigHandler(w http.ResponseWriter, r *http.Request) {
st := struct {
OmsName string // server instance name
DoubleFmt string // format to convert float or double value to string
LoginUrl string // user login URL for UI
LogoutUrl string // user logout URL for UI
AllowUserHome bool // if true then store user settings in home directory
AllowDownload bool // if true then allow download from home/io/download directory
AllowUpload bool // if true then allow upload from home/io/upload directory
AllowMicrodata bool // if true then allow model run microdata
IsJobControl bool // if true then job control enabled
IsArchive bool // if true the archiving is enabled: old data moved out from into archive directory
Env map[string]string // server config environmemt variables for UI
ModelCatalog ModelCatalogConfig // "public" state of model catalog
RunCatalog RunCatalogConfig // "public" state of model run catalog
}{
OmsName: theCfg.omsName,
DoubleFmt: theCfg.doubleFmt,
AllowUserHome: theCfg.isHome,
AllowDownload: theCfg.downloadDir != "",
AllowUpload: theCfg.uploadDir != "",
AllowMicrodata: theCfg.isMicrodata,
IsJobControl: theCfg.isJobControl,
IsArchive: theCfg.isArchive,
Env: theCfg.env,
ModelCatalog: theCatalog.toPublicConfig(),
RunCatalog: *theRunCatalog.toPublicConfig(),
}
jsonResponse(w, r, st)
}
// archiveStateHandler return current state of archive job by reading it from archive-state.json file.
// GET /api/archive/state
func archiveStateHandler(w http.ResponseWriter, r *http.Request) {
// read archive state file and send file content as json response body
bt, err := theArchive.readArchiveState()
if err != nil {
omppLog.Log("Error: unable to read from ", theCfg.archiveStatePath, " ", err)
http.Error(w, "Error: unable to read from "+archiveStateFile, http.StatusInternalServerError)
return // archive state file read error
}
jsonResponseBytes(w, r, bt)
}
// serviceStateHandler return service and model runs state: queue, active runs and run history
// GET /api/service/state
func serviceStateHandler(w http.ResponseWriter, r *http.Request) {
// service state: model run jobs queue, active jobs, history jobs and compute servers state
type cItem struct {
Name string // name of server or cluster
State string // state: start, stop, ready, error, off
TotalRes RunRes // total computational resources (CPU cores and memory)
UsedRes RunRes // resources (CPU cores and memory) used by all oms instances
OwnRes RunRes // resources (CPU cores and memory) used by this instance
ErrorCount int // number of incomplete starts, stops and errors
LastUsedTs int64 // last time for model run (unix milliseconds)
}
st := struct {
IsJobControl bool // if true then job control enabled
JobServiceState // jobs service state: paused, resources usage and limits
Queue []RunJob // list of model run jobs in the queue
Active []RunJob // list of active (currently running) model run jobs
History []historyJobFile // history of model runs
ComputeState []cItem // state of computational servers or clusters
}{
IsJobControl: theCfg.isJobControl,
Queue: []RunJob{},
Active: []RunJob{},
History: []historyJobFile{},
ComputeState: []cItem{},
}
if theCfg.isJobControl {
jsState, qKeys, qJobs, aKeys, aJobs, hKeys, hJobs, cState := theRunCatalog.getRunJobs()
st.JobServiceState = jsState
st.Queue = make([]RunJob, len(qKeys))
for k := range qKeys {
st.Queue[k] = qJobs[k]
st.Queue[k].Env = map[string]string{}
}
st.Active = make([]RunJob, len(aKeys))
for k := range aKeys {
st.Active[k] = aJobs[k]
st.Active[k].Pid = 0
st.Active[k].CmdPath = ""
st.Active[k].LogPath = ""
st.Active[k].Env = map[string]string{}
}
st.History = make([]historyJobFile, len(hKeys))
for k := range hKeys {
st.History[k] = hJobs[k]
}
st.ComputeState = make([]cItem, len(cState))
for k := range cState {
st.ComputeState[k].Name = cState[k].name
st.ComputeState[k].State = cState[k].state
if st.ComputeState[k].State == "" {
st.ComputeState[k].State = "off"
}
st.ComputeState[k].TotalRes = cState[k].totalRes
st.ComputeState[k].UsedRes = cState[k].usedRes
st.ComputeState[k].OwnRes = cState[k].ownRes
st.ComputeState[k].ErrorCount = cState[k].errorCount
st.ComputeState[k].LastUsedTs = cState[k].lastUsedTs
}
}
jsonResponse(w, r, st)
}
// job control state, log file content and run progress
type runJobState struct {
JobStatus string // if not empty then job run status name: success, error, exit
RunJob // job control state: job control file content
RunStatus []db.RunPub // if not empty then run_lst and run_progerss from db
Lines []string // log file content
}
// return empty value of job control state
func emptyRunJobState(submitStamp string) runJobState {
return runJobState{
RunJob: RunJob{
SubmitStamp: submitStamp,
RunRequest: RunRequest{
Opts: map[string]string{},
Env: map[string]string{},
Tables: []string{},
Microdata: struct {
IsToDb bool
IsInternal bool
Entity []struct {
Name string
Attr []string
}
}{
Entity: []struct {
Name string
Attr []string
}{},
},
RunNotes: []struct {
LangCode string
Note string
}{}},
},
RunStatus: []db.RunPub{},
Lines: []string{},
}
}
// jobActiveHandler return active job state, run log file content and, if model run exists in database then also run progress
// GET /api/service/job/active/:job
func jobActiveHandler(w http.ResponseWriter, r *http.Request) {
// url or query parameters: submission stamp
submitStamp := getRequestParam(r, "job")
if submitStamp == "" {
http.Error(w, "Invalid (empty) submission stamp", http.StatusBadRequest)
return
}
// find job state in run catalog
aj, isOk := theRunCatalog.getActiveJobItem(submitStamp)
if !isOk || aj.isError {
jsonResponse(w, r, emptyRunJobState(submitStamp)) // job not found or job control file error
return
}
// get job control state, read log file and run progress, if it is available
isOk, st := getJobState(aj.filePath)
if !isOk {
jsonResponse(w, r, emptyRunJobState(submitStamp)) // unable to read job control file
return
}
jsonResponse(w, r, st) // return final result
}
// jobQueueHandler return queue job state
// GET /api/service/job/queue/:job
func jobQueueHandler(w http.ResponseWriter, r *http.Request) {
// url or query parameters: submission stamp
submitStamp := getRequestParam(r, "job")
if submitStamp == "" {
http.Error(w, "Invalid (empty) submission stamp", http.StatusBadRequest)
return
}
// find job state in run catalog
qj, isOk := theRunCatalog.getQueueJobItem(submitStamp)
if !isOk || qj.isError {
jsonResponse(w, r, emptyRunJobState(submitStamp)) // job not found or job control file error
return
}
// get job control state, log file and run progress are always empty
isOk, st := getJobState(qj.filePath)
if !isOk {
jsonResponse(w, r, emptyRunJobState(submitStamp)) // unable to read job control file
return
}
jsonResponse(w, r, st) // return final result
}
// jobHistoryHandler return history job state, run log file content and, if model run exists in database then also return run progress
// GET /api/service/job/history/:job
func jobHistoryHandler(w http.ResponseWriter, r *http.Request) {
// url or query parameters: submission stamp
submitStamp := getRequestParam(r, "job")
if submitStamp == "" {
http.Error(w, "Invalid (empty) submission stamp", http.StatusBadRequest)
return
}
// find job state in run catalog
hj, isOk := theRunCatalog.getHistoryJobItem(submitStamp)
if !isOk || hj.isError {
jsonResponse(w, r, emptyRunJobState(submitStamp)) // job not found or job control file error
return
}
// get job control state, read log file and run progress, if it is available
isOk, st := getJobState(hj.filePath)
if !isOk {
jsonResponse(w, r, emptyRunJobState(submitStamp)) // unable to read job control file
return
}
st.JobStatus = hj.JobStatus
jsonResponse(w, r, st) // return final result
}
// getJobState returns job control file content, run log file content and, if model run exists in database then also return run progress
// it is clear (set to empty) server-only part of job state: pid, exe path, log path and environment
func getJobState(filePath string) (bool, *runJobState) {
st := emptyRunJobState("")
// read job control file
var jc RunJob
isOk, err := helper.FromJsonFile(filePath, &jc)
if err != nil {
omppLog.Log(err)
}
if !isOk || err != nil {
return false, &st
}
// set job state and clear server-only part of the job state: pid, exe path, log path and environment
st.RunJob = jc
st.Pid = 0
st.CmdPath = ""
st.LogPath = ""
st.Env = map[string]string{}
if len(st.Opts) == 0 {
st.Opts = map[string]string{}
}
if st.Tables == nil {
st.Tables = []string{}
}
if st.Microdata.Entity == nil {
st.Microdata.Entity = []struct {
Name string
Attr []string
}{}
}
if st.RunNotes == nil {
st.RunNotes = []struct {
LangCode string
Note string
}{}
}
// read log file content
if jc.LogPath != "" {
st.Lines, _ = readLogFile(jc.LogPath)
}
// get run progress if model run exist in database
if jc.ModelDigest != "" && jc.RunStamp != "" {
if _, isOk = theCatalog.ModelDicByDigest(jc.ModelDigest); isOk { // if model exist
if rst, isOk := theCatalog.RunStatusList(jc.ModelDigest, jc.RunStamp); isOk { // get run status
st.RunStatus = rst
}
}
}
return true, &st // return final result
}
// jobMoveHandler move job into the specified queue index position.
// PUT /api/service/job/move/:pos/:job
// Top of the queue position is zero, negative position treated as zero.
// If position number exceeds queue length then job moved to the bottom of the queue.
func jobMoveHandler(w http.ResponseWriter, r *http.Request) {
// url or query parameters: position and submission stamp
sp := getRequestParam(r, "pos")
nPos, err := strconv.Atoi(sp)
if sp == "" || err != nil {
http.Error(w, "Invalid (or empty) job queue position", http.StatusBadRequest)
return
}
submitStamp := getRequestParam(r, "job")
if submitStamp == "" {
http.Error(w, "Invalid (empty) submission stamp", http.StatusBadRequest)
return
}
// move job in the queue and rename files in the queue
isOk, fileMoveLst := theRunCatalog.moveJobInQueue(submitStamp, nPos)
for _, fm := range fileMoveLst {
fileMoveAndLog(false, fm[0], fm[1])
}
w.Header().Set("Content-Type", "text/plain")
if !isOk {
w.Header().Set("Content-Location", "service/job/move/false/"+sp+"/"+submitStamp)
return
}
// else: job moved into the specified queue position
w.Header().Set("Content-Location", "service/job/move/true/"+sp+"/"+submitStamp)
}
// jobHistoryDeleteHandler delete only job history json file, it does not delete model run.
// DELETE /api/service/job/delete/history/:job
func jobHistoryDeleteHandler(w http.ResponseWriter, r *http.Request) {
// url or query parameters: submission stamp
submitStamp := getRequestParam(r, "job")
if submitStamp == "" {
http.Error(w, "Invalid (empty) submission stamp", http.StatusBadRequest)
return
}
// find job history in run catalog
hj, isOk := theRunCatalog.getHistoryJobItem(submitStamp)
if isOk {
isOk = fileDeleteAndLog(true, hj.filePath)
if !isOk {
http.Error(w, "Unable to delete job file", http.StatusInternalServerError)
return
}
}
// job history file deleted or job history not found
w.Header().Set("Content-Location", "/api/service/job/delete/history/"+submitStamp)
}