Mercurial > gemma
view pkg/controllers/importqueue.go @ 5591:0011f50cf216 surveysperbottleneckid
Removed no longer used alternative api for surveys/ endpoint.
As bottlenecks in the summary for SR imports are now identified by
their id and no longer by the (not guarantied to be unique!) name,
there is no longer the need to request survey data by the name+date
tuple (which isn't reliable anyway). So the workaround was now
reversed.
author | Sascha Wilde <wilde@sha-bang.de> |
---|---|
date | Wed, 06 Apr 2022 13:30:29 +0200 |
parents | aaa9e658cabd |
children | ceecc205011c |
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 }