Mercurial > gemma
view pkg/controllers/importqueue.go @ 5637:7f9600ab5461
merge
author | Thomas Junk <thomas.junk@intevation.de> |
---|---|
date | Wed, 14 Jun 2023 08:45:11 +0200 |
parents | ceecc205011c |
children | 6270951dda28 1c3bdd114ca1 |
line wrap: on
line source
// This is Free Software under GNU Affero General Public License v >= 3.0 // without warranty, see README.md and license for details. // // SPDX-License-Identifier: AGPL-3.0-or-later // License-Filename: LICENSES/AGPL-3.0.txt // // Copyright (C) 2018 by via donau // – Österreichische Wasserstraßen-Gesellschaft mbH // Software engineering by Intevation GmbH // // Author(s): // * Sascha L. Teichmann <sascha.teichmann@intevation.de> package controllers import ( "context" "database/sql" "encoding/csv" "encoding/json" "fmt" "net/http" "strconv" "strings" "sync" "time" "github.com/gorilla/mux" "github.com/jackc/pgx/pgtype" "gemma.intevation.de/gemma/pkg/auth" "gemma.intevation.de/gemma/pkg/common" "gemma.intevation.de/gemma/pkg/imports" "gemma.intevation.de/gemma/pkg/log" "gemma.intevation.de/gemma/pkg/models" mw "gemma.intevation.de/gemma/pkg/middleware" ) const ( selectImportsCountSQL = ` SELECT count(*) FROM import.imports WHERE ` selectImportsSQL = ` SELECT imports.id AS id, state::varchar, enqueued, changed, kind, username, signer, summary IS NOT NULL AS has_summary, EXISTS(SELECT 1 FROM import.import_logs WHERE kind = 'error'::log_type and import_id = imports.id) AS has_errors, EXISTS(SELECT 1 FROM import.import_logs WHERE kind = 'warn'::log_type and import_id = imports.id) AS has_warnings FROM import.imports WHERE ` // XXX: Consider not only supporting 'sr' for log message parsing. selectExportSQL = ` SELECT imports.id AS id, state::varchar, enqueued, changed, imports.kind, username, (SELECT country FROM users.list_users lu WHERE lu.username = import.imports.username) AS country, signer, EXISTS(SELECT 1 FROM import.import_logs WHERE kind = 'warn'::log_type and import_id = id) AS has_warnings, data, CASE WHEN kind = ANY($1) THEN ARRAY(SELECT msg FROM import.import_logs WHERE import_id = id) ELSE NULL END AS msgs FROM import.imports WHERE ` selectEnqueuedSQL = ` SELECT enqueued FROM import.imports WHERE ` selectImportSummarySQL = ` SELECT summary, enqueued FROM import.imports WHERE id = $1` selectHasNoRunningImportSQL = ` SELECT true FROM import.imports WHERE id = $1 AND state <> 'running'::import_state` selectImportLogsSQL = ` SELECT time, kind::varchar, msg FROM import.import_logs WHERE import_id = $1 ORDER BY time` deleteLogsSQL = ` DELETE FROM import.import_logs WHERE import_id = $1` deleteImportSQL = ` DELETE FROM import.imports WHERE id = $1` ) type filledStmt struct { stmt strings.Builder args []interface{} } func buildFilters(projection string, req *http.Request, args ...interface{}) ( *filledStmt, *filledStmt, *filledStmt, error, ) { var l, a, b filterAnd var noBefore, noAfter bool cond := func(format string, args ...interface{}) { term := &filterTerm{format: format, args: args} l = append(l, term) a = append(a, term) b = append(b, term) } // Always filter review jobs. They are only for internal use. cond(` NOT imports.kind LIKE '%%` + imports.ReviewJobSuffix + `'`) if query := req.FormValue("query"); query != "" { query = "%" + query + "%" cond(` (imports.kind ILIKE $%d OR username ILIKE $%d OR signer ILIKE $%d OR `+ `id IN (SELECT import_id FROM import.import_logs WHERE msg ILIKE $%d)) `, query, query, query, query) } if cc := req.FormValue("cc"); cc != "" { codes := sliceToTextArray(splitUpper(cc)) cond(" username IN "+ "(SELECT username FROM internal.user_profiles "+ "WHERE country = ANY($%d)) ", codes) } if st := req.FormValue("states"); st != "" { states := toTextArray(st, imports.ImportStateNames) cond(" state = ANY($%d) ", states) } if ks := req.FormValue("kinds"); ks != "" { kinds := toTextArray(ks, imports.ImportKindNames()) cond(" imports.kind = ANY($%d) ", kinds) } if idss := req.FormValue("ids"); idss != "" { ids := toInt8Array(idss) cond(" id = ANY($%d) ", ids) } if from := req.FormValue("from"); from != "" { fromTime, err := common.ParseTime(from) if err != nil { return nil, nil, nil, err } l = append(l, buildFilterTerm("enqueued >= $%d", fromTime)) b = append(b, buildFilterTerm("enqueued < $%d", fromTime)) } else { noBefore = true } if to := req.FormValue("to"); to != "" { toTime, err := common.ParseTime(to) if err != nil { return nil, nil, nil, err } l = append(l, buildFilterTerm("enqueued <= $%d", toTime)) a = append(a, buildFilterTerm("enqueued > $%d", toTime)) } else { noAfter = true } switch warn := strings.ToLower(req.FormValue("warnings")); warn { case "1", "t", "true": cond(` EXISTS(SELECT 1 FROM import.import_logs WHERE kind = 'warn'::log_type and import_id = imports.id)`) } switch errors := strings.ToLower(req.FormValue("errors")); errors { case "1", "t", "true": cond(` EXISTS(SELECT 1 FROM import.import_logs WHERE kind = 'error'::log_type and import_id = imports.id)`) } fl := &filledStmt{} fa := &filledStmt{} fb := &filledStmt{} fl.args = append(fl.args, args...) fa.args = append(fa.args, args...) fb.args = append(fb.args, args...) fa.stmt.WriteString(selectEnqueuedSQL) fb.stmt.WriteString(selectEnqueuedSQL) var counting bool if projection != "" { fl.stmt.WriteString(projection) } else { switch count := strings.ToLower(req.FormValue("count")); count { case "1", "t", "true": counting = true fl.stmt.WriteString(selectImportsCountSQL) default: fl.stmt.WriteString(selectImportsSQL) } } if len(l) == 0 { fl.stmt.WriteString(" TRUE ") } else { l.serialize(&fl.stmt, &fl.args) } if len(b) == 0 { fb.stmt.WriteString(" TRUE ") } else { b.serialize(&fb.stmt, &fb.args) } if len(a) == 0 { fa.stmt.WriteString(" TRUE ") } else { a.serialize(&fa.stmt, &fa.args) } if !counting { fl.stmt.WriteString(" ORDER BY enqueued DESC ") fa.stmt.WriteString(" ORDER BY enqueued LIMIT 1") fb.stmt.WriteString(" ORDER BY enqueued DESC LIMIT 1") } if noBefore { fb = nil } if noAfter { fa = nil } return fl, fb, fa, nil } func neighbored(ctx context.Context, conn *sql.Conn, fb *filledStmt) *models.ImportTime { var when time.Time err := conn.QueryRowContext(ctx, fb.stmt.String(), fb.args...).Scan(&when) switch { case err == sql.ErrNoRows: return nil case err != nil: log.Warnf("%v\n", err) return nil } return &models.ImportTime{Time: when.UTC()} } func exportImports(rw http.ResponseWriter, req *http.Request) { type LogLoader interface{ LoadingLogs() bool } var lls []string imports.All(func(k imports.JobKind, jc imports.JobCreator) { if ll, ok := jc.(LogLoader); ok && ll.LoadingLogs() { lls = append(lls, string(k)) } }) var loaders pgtype.TextArray if err := loaders.Set(lls); err != nil { http.Error( rw, fmt.Sprintf("error: %v", err), http.StatusInternalServerError) return } list, _, _, err := buildFilters(selectExportSQL, req, &loaders) if err != nil { http.Error(rw, "error: "+err.Error(), http.StatusBadRequest) return } rw.Header().Add("Content-Type", "text/csv") out := csv.NewWriter(rw) record := []string{ "#id", "#kind", "#enqueued", "#changed", "#user", "#country", "#signer", "#state", "#warnings", "#source", } if err := out.Write(record); err != nil { // Too late for HTTP status message. log.Errorf("%v\n", err) return } start := time.Now() conn := mw.GetDBConn(req) ctx := req.Context() var rows *sql.Rows if rows, err = conn.QueryContext(ctx, list.stmt.String(), list.args...); err != nil { log.Errorf("%v\n", err) return } defer rows.Close() // Extract some meta infos from the import. type Description interface { Description([]string) (string, error) } type dataset struct { id int64 state string enqueued time.Time changed time.Time kind string user string country string signer sql.NullString warnings bool data string msgs pgtype.TextArray } // Log unsupported description interfaces per kind only once. unsupported := make(map[string]bool) store := func(ds *dataset) error { var description string // Do some introspection on the job to be more verbose. if jc := imports.FindJobCreator(imports.JobKind(ds.kind)); jc != nil { job := jc.Create() if err := common.FromJSONString(ds.data, job); err != nil { log.Errorf("%v\n", err) } else if desc, ok := job.(Description); ok { var ms []string if ds.msgs.Status == pgtype.Present { if err := ds.msgs.AssignTo(&ms); err != nil { return err } } if description, err = desc.Description(ms); err != nil { return err } description = strings.Replace(description, ",", "|", -1) } else { if !unsupported[ds.kind] { unsupported[ds.kind] = true log.Debugf("%s: description not supported\n", ds.kind) } } } var signer string if ds.signer.Valid { signer = ds.signer.String } record[0] = strconv.FormatInt(ds.id, 10) record[1] = ds.kind record[2] = ds.enqueued.UTC().Format(common.TimeFormat) record[3] = ds.changed.UTC().Format(common.TimeFormat) record[4] = ds.user record[5] = ds.country record[6] = signer record[7] = ds.state record[8] = strconv.FormatBool(ds.warnings) record[9] = description return out.Write(record) } for rows.Next() { var curr dataset if err := rows.Scan( &curr.id, &curr.state, &curr.enqueued, &curr.changed, &curr.kind, &curr.user, &curr.country, &curr.signer, &curr.warnings, &curr.data, &curr.msgs, ); err != nil { log.Errorf("%v\n", err) return } if err := store(&curr); err != nil { log.Errorf("%v\n", err) return } } if err := rows.Err(); err != nil { log.Errorf("%v\n", err) return } out.Flush() if err := out.Error(); err != nil { log.Errorf("%v\n", err) } log.Debugf("Export took: %v\n", time.Since(start)) } func listImports(req *http.Request) (jr mw.JSONResult, err error) { var list, before, after *filledStmt if list, before, after, err = buildFilters("", req); err != nil { return } ctx := req.Context() conn := mw.JSONConn(req) // Fast path for counting switch count := strings.ToLower(req.FormValue("count")); count { case "1", "t", "true": var count int64 err = conn.QueryRowContext(ctx, list.stmt.String(), list.args...).Scan(&count) switch { case err == sql.ErrNoRows: count, err = 0, nil case err != nil: return } jr = mw.JSONResult{Result: count} return } // Generate the list var rows *sql.Rows if rows, err = conn.QueryContext(ctx, list.stmt.String(), list.args...); err != nil { return } defer rows.Close() imports := make([]*models.Import, 0, 20) var signer sql.NullString for rows.Next() { var it models.Import var enqueued time.Time var changed time.Time if err = rows.Scan( &it.ID, &it.State, &enqueued, &changed, &it.Kind, &it.User, &signer, &it.Summary, &it.Errors, &it.Warnings, ); err != nil { return } if signer.Valid { it.Signer = signer.String } it.Enqueued = models.ImportTime{Time: enqueued.UTC()} it.Changed = models.ImportTime{Time: changed.UTC()} imports = append(imports, &it) } if err = rows.Err(); err != nil { return } var prev, next *models.ImportTime if before != nil { prev = neighbored(ctx, conn, before) } if after != nil { next = neighbored(ctx, conn, after) } jr = mw.JSONResult{ Result: struct { Prev *models.ImportTime `json:"prev,omitempty"` Next *models.ImportTime `json:"next,omitempty"` Imports []*models.Import `json:"imports"` }{ Imports: imports, Prev: prev, Next: next, }, } return } func importLogs(req *http.Request) (jr mw.JSONResult, err error) { ctx := req.Context() id, _ := strconv.ParseInt(mux.Vars(req)["id"], 10, 64) conn := mw.JSONConn(req) // Check if he have such a import job first. var summary sql.NullString var enqueued time.Time err = conn.QueryRowContext(ctx, selectImportSummarySQL, id).Scan( &summary, &enqueued, ) switch { case err == sql.ErrNoRows: err = mw.JSONError{ Code: http.StatusNotFound, Message: fmt.Sprintf("Cannot find import #%d.", id), } return case err != nil: return } enqueued = enqueued.UTC() var sum interface{} if summary.Valid { if err = json.NewDecoder( strings.NewReader(summary.String)).Decode(&sum); err != nil { return } } // We have it -> generate log entries. var rows *sql.Rows rows, err = conn.QueryContext(ctx, selectImportLogsSQL, id) if err != nil { return } defer rows.Close() entries := make([]*models.ImportLogEntry, 0, 10) for rows.Next() { var entry models.ImportLogEntry var t time.Time if err = rows.Scan(&t, &entry.Kind, &entry.Message); err != nil { return } entry.Time = models.ImportTime{Time: t.UTC()} entries = append(entries, &entry) } if err = rows.Err(); err != nil { return } jr = mw.JSONResult{ Result: struct { Enqueued models.ImportTime `json:"enqueued"` Summary interface{} `json:"summary,omitempty"` Entries []*models.ImportLogEntry `json:"entries"` }{ Enqueued: models.ImportTime{Time: enqueued}, Summary: sum, Entries: entries, }, } return } func deleteImport(req *http.Request) (jr mw.JSONResult, err error) { ctx := req.Context() id, _ := strconv.ParseInt(mux.Vars(req)["id"], 10, 64) var tx *sql.Tx tx, err = mw.JSONConn(req).BeginTx(ctx, nil) if err != nil { return } defer tx.Rollback() // Check if he have such a import job first. var dummy bool err = tx.QueryRowContext(ctx, selectHasNoRunningImportSQL, id).Scan(&dummy) switch { case err == sql.ErrNoRows: err = mw.JSONError{ Code: http.StatusNotFound, Message: fmt.Sprintf("Cannot find import #%d.", id), } return case err != nil: return } if _, err = tx.ExecContext(ctx, deleteLogsSQL, id); err != nil { return } if _, err = tx.ExecContext(ctx, deleteImportSQL, id); err != nil { return } if err = tx.Commit(); err != nil { return } jr = mw.JSONResult{Code: http.StatusNoContent} return } func reviewImports(req *http.Request) (mw.JSONResult, error) { rs := *mw.JSONInput(req).(*[]models.Review) type reviewResult struct { ID int64 `json:"id"` Message string `json:"message,omitempty"` Error string `json:"error,omitempty"` } results := make([]reviewResult, len(rs)) for i := range results { results[i].ID = rs[i].ID results[i].Message = fmt.Sprintf("Finalizing import #%d in progress.", rs[i].ID) } var wg sync.WaitGroup var mu sync.Mutex for i := range rs { wg.Add(1) go func(idx int) { defer wg.Done() rev := &rs[idx] msg, err := decideImport(req, rev.ID, string(rev.State)) mu.Lock() if err != nil { results[idx].Error = err.Error() } results[idx].Message = msg mu.Unlock() }(i) } done := make(chan struct{}) go func() { defer close(done) wg.Wait() }() select { case <-time.After(5 * time.Second): case <-done: } out := make([]reviewResult, len(rs)) mu.Lock() copy(out, results) mu.Unlock() return mw.JSONResult{Result: out}, nil } func reviewImport(req *http.Request) (jr mw.JSONResult, err error) { vars := mux.Vars(req) id, _ := strconv.ParseInt(vars["id"], 10, 64) state := vars["state"] var msg string if msg, err = decideImport(req, id, state); err != nil { return } result := struct { Message string `json:"message"` }{ Message: msg, } jr = mw.JSONResult{Result: &result} return } func decideImport( req *http.Request, id int64, state string, ) (message string, err error) { session, _ := auth.GetSession(req) reviewer := session.User ctx := req.Context() accepted := state == "accepted" switch err = imports.DecideImport(ctx, id, accepted, reviewer); { case err == imports.ErrRetrying: return "", fmt.Errorf("finalizing import #%d is re-scheduled", id) case err != nil: return "", err } return fmt.Sprintf("Import #%d is %s.", id, state), nil }