;;======================================================================
;; Copyright 2017, Matthew Welland.
;;
;; This file is part of Megatest.
;;
;; Megatest is free software: you can redistribute it and/or modify
;; it under the terms of the GNU General Public License as published by
;; the Free Software Foundation, either version 3 of the License, or
;; (at your option) any later version.
;;
;; Megatest is distributed in the hope that it will be useful,
;; but WITHOUT ANY WARRANTY; without even the implied warranty of
;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
;; GNU General Public License for more details.
;;
;; You should have received a copy of the GNU General Public License
;; along with Megatest. If not, see <http://www.gnu.org/licenses/>.
;;======================================================================
(declare (unit rmtmod))
(declare (uses apimod))
(declare (uses commonmod))
(declare (uses configfmod))
(declare (uses dbmod))
(declare (uses debugprint))
(declare (uses itemsmod))
(declare (uses mtargs))
(declare (uses mtver))
(declare (uses pgdb))
(declare (uses portloggermod))
(declare (uses servermod))
(declare (uses tasksmod))
(module rmtmod
*
(import scheme
chicken.base
chicken.condition
chicken.file
chicken.file.posix
chicken.format
chicken.io
chicken.pathname
chicken.port
chicken.pretty-print
chicken.process
chicken.process-context
chicken.process-context.posix
chicken.sort
chicken.string
;; chicken.tcp
chicken.random
chicken.time
chicken.time.posix
(prefix sqlite3 sqlite3:)
directory-utils
;; http-client
;; intarweb
matchable
md5
message-digest
nng ;; nanomsg
(prefix base64 base64:)
(prefix sqlite3 sqlite3:)
regex
s11n
;; spiffy
;; spiffy-directory-listing
;; spiffy-request-vars
srfi-1
srfi-13
srfi-18
srfi-69
stack
system-information
;; tcp6
typed-records
uri-common
z3
apimod
commonmod
configfmod
dbmod
debugprint
itemsmod
mtver
pgdb
pkts
portloggermod
(prefix mtargs args:)
servermod
stml2
tasksmod
)
(defstruct alldat
(areapath #f)
(ulexdat #f)
)
;; (require-extension (srfi 18) extras tcp s11n)
;;
;;
;; (use srfi-1 posix regex regex-case srfi-69 hostinfo md5 message-digest posix-extras)
;;
;; (use spiffy uri-common intarweb http-client spiffy-request-vars intarweb spiffy-directory-listing)
;;
;; Configurations for server
;; (tcp-buffer-size 2048)
;; (max-connections 2048)
;; info about me as a server
;;
(defstruct servdat
(host #f)
(port #f)
(uuid #f)
(rep #f)
(dbfile #f)
(api-url #f)
(api-uri #f)
(api-req #f)
(status 'starting)
(trynum 0) ;; count the number of ports we've tried
)
(define (servdat->url sdat)
(conc (servdat-host sdat)":"(servdat-port sdat)))
;; generate entries for ~/.megatestrc with the following
;;
;; grep define ../rmt.scm | grep rmt: |perl -pi -e 's/\(define\s+\((\S+)\W.*$/\1/'|sort -u
(defstruct rmt:remote
(conns (make-hash-table)) ;; apath/dbname => rmt:conn
)
(defstruct rmt:conn
(apath #f)
(dbname #f)
(fullname #f)
(hostport #f)
(ipaddr #f)
(port #f)
(srvpkt #f)
(srvkey #f)
(lastmsg 0)
(expires 0)
(inport #f)
(outport #f))
(define *srvpktspec*
`((server (host . h)
(port . p)
(servkey . k)
(pid . i)
(ipaddr . a)
(dbpath . d))))
;;======================================================================
;; S U P P O R T F U N C T I O N S
;;======================================================================
;; replaces *runremote*
(define *rmt:remote* (make-rmt:remote))
;; -> http://abc.com:900/<entrypoint>
;;
(define (rmt:conn->uri conn entrypoint)
(conc "http://"(rmt:conn-ipaddr conn)":"(rmt:conn-port conn)"/"entrypoint))
;; set up the api proc, seems like there should be a better place for this?
(define api-proc (make-parameter conc))
(api-proc api:process-request)
;; do we have a connection to apath dbname and
;; is it not expired? then return it
;;
;; else setup a connection
;;
;; if that fails, return '(#f "some reason") ;; NB// convert to raising an exception
;;
(define (rmt:get-conn remote apath dbname)
(let* ((fullname (db:dbname->path apath dbname)) ;; we'll switch to full name later
(conn (hash-table-ref/default (rmt:remote-conns remote) dbname #f)))
(if (and conn
(< (current-seconds) (rmt:conn-expires conn)))
conn
#f)))
(define (rmt:find-main-server apath dbname)
(let* ((pktsdir (get-pkts-dir apath))
(all-srvpkts (get-all-server-pkts pktsdir *srvpktspec*))
;; (dbpath (conc apath "/" dbname))
(viable-srvs (get-viable-servers all-srvpkts dbname)))
(get-the-server apath viable-srvs)))
;; looks for a connection to main
;; connections for other servers happens by requesting from main
;;
;; TODO: This is unnecessarily re-creating the record in the hash table
;;
(define (rmt:open-main-connection remote apath)
(let* ((dbname (db:run-id->dbname #f))
(the-srv (rmt:find-main-server apath dbname))
(start-main-srv (lambda ()
;; srv not ready, delay a little and try again
(api:run-server-process apath dbname)
(thread-sleep! 4)
(rmt:open-main-connection remote apath) ;; TODO: Add limit to number of tries
)))
(if the-srv ;; yes, we have a server, now try connecting to it
(let* ((srv-addr (server-address the-srv))
(ipaddr (alist-ref 'ipaddr the-srv))
(port (alist-ref 'port the-srv))
(srvkey (alist-ref 'servkey the-srv))
(fullpath (db:dbname->path apath dbname))
(srvready (server-ready? ipaddr port srvkey)))
(if srvready
(begin
(hash-table-set! (rmt:remote-conns remote)
dbname ;; fullpath ;; yes, I'd prefer it to be fullpath - FIXME later
(make-rmt:conn
apath: apath
dbname: dbname
fullname: fullpath
hostport: srv-addr
ipaddr: ipaddr
port: port
srvpkt: the-srv
srvkey: srvkey ;; generated by rmt:get-signature on the server side
lastmsg: (current-seconds)
expires: (+ (current-seconds) 60) ;; this needs to be gathered during the ping
))
#t)
(start-main-srv)))
(start-main-srv))))
;; NB// remote is a rmt:remote struct
;;
(define (rmt:general-open-connection remote apath dbname #!key (num-tries 5))
(let* ((mdbname (db:run-id->dbname #f))
(mconn (rmt:get-conn remote apath mdbname)))
(cond
((or (not mconn) ;; no channel open to main?
(< (rmt:conn-expires mconn)(+ (current-seconds) 2))) ;; restablish connection if less than 2 seconds on the lease
(rmt:open-main-connection remote apath)
(rmt:general-open-connection remote apath mdbname))
((not (rmt:get-conn remote apath dbname)) ;; no channel open to dbname?
(let* ((res (rmt:send-receive-real remote apath mdbname 'get-server `(,apath ,dbname))))
(case res
((server-started)
(if (> num-tries 0)
(begin
(thread-sleep! 2)
(rmt:general-open-connection remote apath dbname num-tries: (- num-tries 1)))
(begin
(debug:print-error 0 *default-log-port* "Failed to start servers needed or open channel to "apath", "dbname)
(exit 1))))
(else
(if (list? res) ;; server has been registered and the info was returned. pass it on.
(begin ;; ("192.168.0.9" 53817
;; "5e34239f48e8973b3813221e54701a01" "24310"
;; "192.168.0.9"
;; "/home/matt/data/megatest/tests/simplerun"
;; ".db/1.db")
(match
res
((host port servkey pid ipaddr apath dbname)
(debug:print-info 0 *default-log-port* "got "res)
(hash-table-set! (rmt:remote-conns remote)
dbname
(make-rmt:conn
apath: apath
dbname: dbname
hostport: (conc host":"port)
ipaddr: ipaddr
port: port
srvkey: servkey
lastmsg: (current-seconds)
expires: (+ (current-seconds) 60))))
(else
(debug:print-info 0 *default-log-port* "return data from starting server did not match host port servkey pid ipaddr apath dbname " res)))
res)
(begin
(debug:print-info 0 *default-log-port* "Unexpected result: " res)
res))))))
)))
;;======================================================================
;; FOR DEBUGGING SET TO #t
(define *localmode* #t)
(define *dbstruct* (make-dbr:dbstruct))
;; Defaults to current area
;;
(define (rmt:send-receive cmd rid params #!key (attemptnum 1)(area-dat #f))
(if (not *rmt:remote*)(set! *rmt:remote* (make-rmt:remote)))
(let* ((apath *toppath*)
(conns *rmt:remote*)
(dbname (db:run-id->dbname rid)))
(if *localmode*
(let* ((dbdat (dbr:dbstruct-get-dbdat *dbstruct* dbname))
(indat `((cmd . ,cmd)(params . ,params))))
(api:process-request *dbstruct* indat))
(begin
(rmt:general-open-connection conns apath dbname)
(rmt:send-receive-real conns apath dbname cmd params)))))
#;(define (rmt:send-receive-setup conn)
(if (not (rmt:conn-inport conn))
(let-values (((i o) (tcp-connect (rmt:conn-ipaddr conn)
(rmt:conn-port conn))))
(rmt:conn-inport-set! conn i)
(rmt:conn-outport-set! conn o))))
;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
;; sometime in the future
;;
(define (rmt:send-receive-real remote apath dbname cmd params)
(let* ((conn (rmt:get-conn remote apath dbname)))
(assert conn "FATAL: rmt:send-receive-real called without the needed channels opened")
(let* ((key #f)
(host (rmt:conn-ipaddr conn))
(port (rmt:conn-port conn))
(payload `((cmd . ,cmd)
(key . ,(rmt:conn-srvkey conn))
(params . ,params)))
(res (open-send-receive-nn (conc host":"port)
(sexpr->string payload))))
(string->sexpr res))))
;; db is at apath/.db/dbname, rid is an intermediary solution and will be removed
;; sometime in the future.
;;
;; Purpose - call the main.db server and request a server be started
;; for the given area path and dbname
;;
;; (define (rmt:send-receive-server-start remote apath dbname)
;; (let* ((conn (rmt:get-conn remote apath dbname)))
;; (assert conn "FATAL: Unable to connect to db "apath"/"dbname)
;; #;(let* ((res (with-input-from-request
;; (rmt:conn->uri conn "api")
;; `((params . (,apath ,dbname)))
;; read-string)))
;; (string->sexpr res))))
(define (rmt:print-db-stats)
(let ((fmtstr "~40a~7-d~9-d~20,2-f")) ;; "~20,2-f"
(debug:print 18 *default-log-port* "DB Stats, "(seconds->year-week/day-time (current-seconds))"\n=====================")
(debug:print 18 *default-log-port* (format #f "~40a~8a~10a~10a" "Cmd" "Count" "TotTime" "Avg"))
(for-each (lambda (cmd)
(let ((cmd-dat (hash-table-ref *db-stats* cmd)))
(debug:print 18 *default-log-port* (format #f fmtstr cmd (vector-ref cmd-dat 0) (vector-ref cmd-dat 1) (/ (vector-ref cmd-dat 1)(vector-ref cmd-dat 0))))))
(sort (hash-table-keys *db-stats*)
(lambda (a b)
(> (vector-ref (hash-table-ref *db-stats* a) 0)
(vector-ref (hash-table-ref *db-stats* b) 0)))))))
(define (rmt:get-max-query-average run-id)
(mutex-lock! *db-stats-mutex*)
(let* ((runkey (conc "run-id=" run-id " "))
(cmds (filter (lambda (x)
(substring-index runkey x))
(hash-table-keys *db-stats*)))
(res (if (null? cmds)
(cons 'none 0)
(let loop ((cmd (car cmds))
(tal (cdr cmds))
(max-cmd (car cmds))
(res 0))
(let* ((cmd-dat (hash-table-ref *db-stats* cmd))
(tot (vector-ref cmd-dat 0))
(curravg (/ (vector-ref cmd-dat 1) (vector-ref cmd-dat 0))) ;; count is never zero by construction
(currmax (max res curravg))
(newmax-cmd (if (> curravg res) cmd max-cmd)))
(if (null? tal)
(if (> tot 10)
(cons newmax-cmd currmax)
(cons 'none 0))
(loop (car tal)(cdr tal) newmax-cmd currmax)))))))
(mutex-unlock! *db-stats-mutex*)
res))
;;======================================================================
;;
;; A C T U A L A P I C A L L S
;;
;;======================================================================
;;======================================================================
;; S E R V E R
;;======================================================================
(define (rmt:kill-server run-id)
(rmt:send-receive 'kill-server run-id (list run-id)))
(define (rmt:start-server run-id)
(rmt:send-receive 'start-server 0 (list run-id)))
;;======================================================================
;; M I S C
;;======================================================================
(define (rmt:login run-id)
(rmt:send-receive 'login run-id (list *toppath* megatest-version *my-signature*)))
;; rmt:login-no-auto-client-setup
;; rmt:send-receive-no-auto-client-setup
;; hand off a call to one of the db:queries statements
;; added run-id to make looking up the correct db possible
;;
(define (rmt:general-call stmtname run-id . params)
(rmt:send-receive 'general-call run-id (append (list stmtname run-id) params)))
;; given a hostname, return a pair of cpu load and update time representing latest intelligence from tests running on that host
(define (rmt:get-latest-host-load hostname)
(rmt:send-receive 'get-latest-host-load 0 (list hostname)))
(define (rmt:sdb-qry qry val run-id)
;; add caching if qry is 'getid or 'getstr
(rmt:send-receive 'sdb-qry run-id (list qry val)))
;; NOT COMPLETED
(define (rmt:runtests user run-id testpatt params)
(rmt:send-receive 'runtests run-id testpatt))
(define (rmt:get-run-record-ids target run keynames test-patt)
(rmt:send-receive 'get-run-record-ids #f (list target run keynames test-patt)))
(define (rmt:get-changed-record-ids since-time)
(rmt:send-receive 'get-changed-record-ids #f (list since-time)) )
(define (rmt:drop-all-triggers)
(rmt:send-receive 'drop-all-triggers #f '()))
(define (rmt:create-all-triggers)
(rmt:send-receive 'create-all-triggers #f '()))
;;======================================================================
;; T E S T M E T A
;;======================================================================
(define (rmt:get-tests-tags)
(rmt:send-receive 'get-tests-tags #f '()))
;;======================================================================
;; K E Y S
;;======================================================================
;; These require run-id because the values come from the run!
;;
(define (rmt:get-key-val-pairs run-id)
(rmt:send-receive 'get-key-val-pairs run-id (list run-id)))
(define (rmt:get-keys)
(if *db-keys* *db-keys*
(let ((res (rmt:send-receive 'get-keys #f '())))
(set! *db-keys* res)
res)))
(define (rmt:get-keys-write) ;; dummy query to force server start
(let ((res (rmt:send-receive 'get-keys-write #f '())))
(set! *db-keys* res)
res))
;; we don't reuse run-id's (except possibly *after* a db cleanup) so it is safe
;; to cache the resuls in a hash
;;
(define (rmt:get-key-vals run-id)
(or (hash-table-ref/default *keyvals* run-id #f)
(let ((res (rmt:send-receive 'get-key-vals #f (list run-id))))
(hash-table-set! *keyvals* run-id res)
res)))
(define (rmt:get-targets)
(rmt:send-receive 'get-targets #f '()))
(define (rmt:get-target run-id)
(rmt:send-receive 'get-target run-id (list run-id)))
(define (rmt:get-run-times runpatt targetpatt)
(rmt:send-receive 'get-run-times #f (list runpatt targetpatt )))
;;======================================================================
;; T E S T S
;;======================================================================
;; Just some syntatic sugar
(define (rmt:register-test run-id test-name item-path)
(rmt:general-call 'register-test run-id run-id test-name item-path))
(define (rmt:get-test-id run-id testname item-path)
(rmt:send-receive 'get-test-id run-id (list run-id testname item-path)))
;; run-id is NOT used
;;
(define (rmt:get-test-info-by-id run-id test-id)
(if (number? test-id)
(rmt:send-receive 'get-test-info-by-id run-id (list run-id test-id))
(begin
(debug:print 0 *default-log-port* "WARNING: Bad data handed to rmt:get-test-info-by-id run-id=" run-id ", test-id=" test-id)
(print-call-chain (current-error-port))
#f)))
(define (rmt:test-get-rundir-from-test-id run-id test-id)
(rmt:send-receive 'test-get-rundir-from-test-id run-id (list run-id test-id)))
;; (define (rmt:open-test-db-by-test-id run-id test-id #!key (work-area #f))
;; (let* ((test-path (if (string? work-area)
;; work-area
;; (rmt:test-get-rundir-from-test-id run-id test-id))))
;; (debug:print 3 *default-log-port* "TEST PATH: " test-path)
;; (open-test-db test-path)))
;; WARNING: This currently bypasses the transaction wrapped writes system
(define (rmt:test-set-state-status-by-id run-id test-id newstate newstatus newcomment)
(rmt:send-receive 'test-set-state-status-by-id run-id (list run-id test-id newstate newstatus newcomment)))
(define (rmt:set-tests-state-status run-id testnames currstate currstatus newstate newstatus)
(rmt:send-receive 'set-tests-state-status run-id (list run-id testnames currstate currstatus newstate newstatus)))
(define (rmt:get-tests-for-run run-id testpatt states statuses offset limit not-in sort-by sort-order qryvals last-update mode)
;; (if (number? run-id)
(rmt:send-receive 'get-tests-for-run run-id (list run-id testpatt states statuses offset limit not-in sort-by sort-order qryvals last-update mode)))
;; (begin
;; (debug:print-error 0 *default-log-port* "rmt:get-tests-for-run called with bad run-id=" run-id)
;; (print-call-chain (current-error-port))
;; '())))
(define (rmt:get-tests-for-run-state-status run-id testpatt last-update)
(rmt:send-receive 'get-tests-for-run-state-status run-id (list run-id testpatt last-update)))
;; get stuff via synchash
(define (rmt:synchash-get run-id proc synckey keynum params)
(rmt:send-receive 'synchash-get run-id (list run-id proc synckey keynum params)))
(define (rmt:get-tests-for-run-mindata run-id testpatt states status not-in)
(rmt:send-receive 'get-tests-for-run-mindata run-id (list run-id testpatt states status not-in)))
;; IDEA: Threadify these - they spend a lot of time waiting ...
;;
(define (rmt:get-tests-for-runs-mindata run-ids testpatt states status not-in)
(let ((multi-run-mutex (make-mutex))
(run-id-list (if run-ids
run-ids
(rmt:get-all-run-ids)))
(result '()))
(if (null? run-id-list)
'()
(let loop ((hed (car run-id-list))
(tal (cdr run-id-list))
(threads '()))
(if (> (length threads) 5)
(loop hed tal (filter (lambda (th)(not (member (thread-state th) '(terminated dead)))) threads))
(let* ((newthread (make-thread
(lambda ()
(let ((res (rmt:send-receive 'get-tests-for-run-mindata hed (list hed testpatt states status not-in))))
(if (list? res)
(begin
(mutex-lock! multi-run-mutex)
(set! result (append result res))
(mutex-unlock! multi-run-mutex))
(debug:print-error 0 *default-log-port* "get-tests-for-run-mindata failed for run-id " hed ", testpatt " testpatt ", states " states ", status " status ", not-in " not-in))))
(conc "multi-run-thread for run-id " hed)))
(newthreads (cons newthread threads)))
(thread-start! newthread)
(thread-sleep! 0.054) ;; give that thread some time to start
(if (null? tal)
newthreads
(loop (car tal)(cdr tal) newthreads))))))
result))
;; ;; IDEA: Threadify these - they spend a lot of time waiting ...
;; ;;
;; (define (rmt:get-tests-for-runs-mindata run-ids testpatt states status not-in)
;; (let ((run-id-list (if run-ids
;; run-ids
;; (rmt:get-all-run-ids))))
;; (apply append (map (lambda (run-id)
;; (rmt:send-receive 'get-tests-for-run-mindata run-id (list run-ids testpatt states status not-in)))
;; run-id-list))))
(define (rmt:delete-test-records run-id test-id)
(rmt:send-receive 'delete-test-records run-id (list run-id test-id)))
(define (rmt:test-set-state-status run-id test-id state status msg)
(rmt:send-receive 'test-set-state-status run-id (list run-id test-id state status msg)))
(define (rmt:test-toplevel-num-items run-id test-name)
(rmt:send-receive 'test-toplevel-num-items run-id (list run-id test-name)))
;; (define (rmt:get-previous-test-run-record run-id test-name item-path)
;; (rmt:send-receive 'get-previous-test-run-record run-id (list run-id test-name item-path)))
(define (rmt:get-matching-previous-test-run-records run-id test-name item-path)
(rmt:send-receive 'get-matching-previous-test-run-records run-id (list run-id test-name item-path)))
(define (rmt:test-get-logfile-info run-id test-name)
(rmt:send-receive 'test-get-logfile-info run-id (list run-id test-name)))
(define (rmt:test-get-records-for-index-file run-id test-name)
(rmt:send-receive 'test-get-records-for-index-file run-id (list run-id test-name)))
(define (rmt:get-testinfo-state-status run-id test-id)
(rmt:send-receive 'get-testinfo-state-status run-id (list run-id test-id)))
(define (rmt:test-set-log! run-id test-id logf)
(if (string? logf)(rmt:general-call 'test-set-log run-id logf test-id)))
(define (rmt:test-set-top-process-pid run-id test-id pid)
(rmt:send-receive 'test-set-top-process-pid run-id (list run-id test-id pid)))
(define (rmt:test-get-top-process-pid run-id test-id)
(rmt:send-receive 'test-get-top-process-pid run-id (list run-id test-id)))
(define (rmt:get-run-ids-matching-target keynames target res runname testpatt statepatt statuspatt)
(rmt:send-receive 'get-run-ids-matching-target #f (list keynames target res runname testpatt statepatt statuspatt)))
;; NOTE: This will open and access ALL run databases.
;;
(define (rmt:test-get-paths-matching-keynames-target-new keynames target res testpatt statepatt statuspatt runname)
(let ((run-ids (rmt:get-run-ids-matching-target keynames target res runname testpatt statepatt statuspatt)))
(apply append
(map (lambda (run-id)
(rmt:send-receive 'test-get-paths-matching-keynames-target-new run-id (list run-id keynames target res testpatt statepatt statuspatt runname)))
run-ids))))
(define (rmt:get-prereqs-not-met run-id waitons ref-test-name ref-item-path #!key (mode '(normal))(itemmaps #f))
(rmt:send-receive 'get-prereqs-not-met run-id (list run-id waitons ref-test-name ref-item-path mode itemmaps)))
(define (rmt:get-count-tests-running-for-run-id run-id)
(rmt:send-receive 'get-count-tests-running-for-run-id run-id (list run-id)))
(define (rmt:get-not-completed-cnt run-id)
(rmt:send-receive 'get-not-completed-cnt run-id (list run-id)))
;; Statistical queries
(define (rmt:get-count-tests-running run-id)
(rmt:send-receive 'get-count-tests-running run-id (list run-id)))
(define (rmt:get-count-tests-running-for-testname run-id testname)
(rmt:send-receive 'get-count-tests-running-for-testname run-id (list run-id testname)))
(define (rmt:get-count-tests-running-in-jobgroup run-id jobgroup)
(rmt:send-receive 'get-count-tests-running-in-jobgroup run-id (list run-id jobgroup)))
;; state and status are extra hints not usually used in the calculation
;;
(define (rmt:set-state-status-and-roll-up-items run-id test-name item-path state status comment)
(rmt:send-receive 'set-state-status-and-roll-up-items run-id (list run-id test-name item-path state status comment)))
(define (rmt:set-state-status-and-roll-up-run run-id state status)
(rmt:send-receive 'set-state-status-and-roll-up-run run-id (list run-id state status)))
(define (rmt:update-pass-fail-counts run-id test-name)
(rmt:general-call 'update-pass-fail-counts run-id test-name test-name test-name))
(define (rmt:top-test-set-per-pf-counts run-id test-name)
(rmt:send-receive 'top-test-set-per-pf-counts run-id (list run-id test-name)))
(define (rmt:get-raw-run-stats run-id)
(rmt:send-receive 'get-raw-run-stats run-id (list run-id)))
(define (rmt:get-test-times runname target)
(rmt:send-receive 'get-test-times #f (list runname target )))
;;======================================================================
;; R U N S
;;======================================================================
(define (rmt:get-run-info run-id)
(rmt:send-receive 'get-run-info run-id (list run-id)))
(define (rmt:get-num-runs runpatt)
(rmt:send-receive 'get-num-runs #f (list runpatt)))
(define (rmt:get-runs-cnt-by-patt runpatt targetpatt keys)
(rmt:send-receive 'get-runs-cnt-by-patt #f (list runpatt targetpatt keys)))
;; Use the special run-id == #f scenario here since there is no run yet
(define (rmt:register-run keyvals runname state status user contour)
;; first register in main.db (thus the #f)
(let* ((run-id (rmt:send-receive 'register-run #f (list keyvals runname state status user contour))))
;; now register in the run db itself
;; NEED A RECORD INSERT INCLUDING SETTING id
(rmt:send-receive 'insert-run run-id (list run-id keyvals runname state status user contour))
run-id))
(define (rmt:get-run-name-from-id run-id)
(rmt:send-receive 'get-run-name-from-id run-id (list run-id)))
(define (rmt:delete-run run-id)
(rmt:send-receive 'delete-run run-id (list run-id)))
(define (rmt:update-run-stats run-id stats)
(rmt:send-receive 'update-run-stats #f (list run-id stats)))
(define (rmt:delete-old-deleted-test-records)
(rmt:send-receive 'delete-old-deleted-test-records #f '()))
(define (rmt:get-runs runpatt count offset keypatts)
(rmt:send-receive 'get-runs #f (list runpatt count offset keypatts)))
(define (rmt:simple-get-runs runpatt count offset target last-update)
(rmt:send-receive 'simple-get-runs #f (list runpatt count offset target last-update)))
(define (rmt:get-all-run-ids)
(rmt:send-receive 'get-all-run-ids #f '()))
(define (rmt:get-prev-run-ids run-id)
(rmt:send-receive 'get-prev-run-ids #f (list run-id)))
(define (rmt:lock/unlock-run run-id lock unlock user)
(rmt:send-receive 'lock/unlock-run #f (list run-id lock unlock user)))
;; set/get status
(define (rmt:get-run-status run-id)
(rmt:send-receive 'get-run-status #f (list run-id)))
(define (rmt:get-run-state run-id)
(rmt:send-receive 'get-run-state #f (list run-id)))
(define (rmt:set-run-status run-id run-status #!key (msg #f))
(rmt:send-receive 'set-run-status #f (list run-id run-status msg)))
(define (rmt:set-run-state-status run-id state status )
(rmt:send-receive 'set-run-state-status #f (list run-id state status)))
(define (rmt:update-tesdata-on-repilcate-db old-lt new-lt)
(rmt:send-receive 'update-tesdata-on-repilcate-db #f (list old-lt new-lt)))
(define (rmt:update-run-event_time run-id)
(rmt:send-receive 'update-run-event_time #f (list run-id)))
(define (rmt:get-runs-by-patt keys runnamepatt targpatt offset limit fields last-runs-update #!key (sort-order "asc")) ;; fields of #f uses default
(rmt:send-receive 'get-runs-by-patt #f (list keys runnamepatt targpatt offset limit fields last-runs-update sort-order)))
(define (rmt:find-and-mark-incomplete run-id ovr-deadtime)
;; (if (rmt:send-receive 'have-incompletes? run-id (list run-id ovr-deadtime))
(rmt:send-receive 'mark-incomplete run-id (list run-id ovr-deadtime))
) ;; )
(define (rmt:get-main-run-stats run-id)
(rmt:send-receive 'get-main-run-stats #f (list run-id)))
(define (rmt:get-var run-id varname)
(rmt:send-receive 'get-var run-id (list run-id varname)))
(define (rmt:del-var run-id varname)
(rmt:send-receive 'del-var run-id (list run-id varname)))
(define (rmt:set-var run-id varname value)
(rmt:send-receive 'set-var run-id (list run-id varname value)))
(define (rmt:inc-var run-id varname)
(rmt:send-receive 'inc-var #f (list run-id varname)))
(define (rmt:dec-var run-id varname)
(rmt:send-receive 'dec-var run-id (list run-id varname)))
(define (rmt:add-var run-id varname value)
(rmt:send-receive 'add-var run-id (list run-id varname value)))
;;======================================================================
;; M U L T I R U N Q U E R I E S
;;======================================================================
;; Need to move this to multi-run section and make associated changes
(define (rmt:find-and-mark-incomplete-all-runs #!key (ovr-deadtime #f))
(let ((run-ids (rmt:get-all-run-ids)))
(for-each (lambda (run-id)
(rmt:find-and-mark-incomplete run-id ovr-deadtime))
run-ids)))
;; get the previous record for when this test was run where all keys match but runname
;; returns #f if no such test found, returns a single test record if found
;;
;; Run this at the client end since we have to connect to multiple run-id dbs
;;
(define (rmt:get-previous-test-run-record run-id test-name item-path)
(let* ((keyvals (rmt:get-key-val-pairs run-id))
(keys (rmt:get-keys))
(selstr (string-intersperse keys ","))
(qrystr (string-intersperse (map (lambda (x)(conc x "=?")) keys) " AND ")))
(if (not keyvals)
#f
(let ((prev-run-ids (rmt:get-prev-run-ids run-id)))
;; for each run starting with the most recent look to see if there is a matching test
;; if found then return that matching test record
(debug:print 4 *default-log-port* "selstr: " selstr ", qrystr: " qrystr ", keyvals: " keyvals ", previous run ids found: " prev-run-ids)
(if (null? prev-run-ids) #f
(let loop ((hed (car prev-run-ids))
(tal (cdr prev-run-ids)))
(let ((results (rmt:get-tests-for-run hed (conc test-name "/" item-path) '() '() ;; run-id testpatt states statuses
#f #f #f ;; offset limit not-in hide/not-hide
#f #f #f #f 'normal))) ;; sort-by sort-order qryvals last-update mode
(debug:print 4 *default-log-port* "Got tests for run-id " run-id ", test-name " test-name ", item-path " item-path ": " results)
(if (and (null? results)
(not (null? tal)))
(loop (car tal)(cdr tal))
(if (null? results) #f
(car results))))))))))
(define (rmt:get-run-stats)
(rmt:send-receive 'get-run-stats #f '()))
;;======================================================================
;; S T E P S
;;======================================================================
;; Getting steps is more complicated.
;;
;; If given work area
;; 1. Find the testdat.db file
;; 2. Open the testdat.db file and do the query
;; If not given the work area
;; 1. Do a remote call to get the test path
;; 2. Continue as above
;;
;;(define (rmt:get-steps-for-test run-id test-id)
;; (rmt:send-receive 'get-steps-data run-id (list test-id)))
(define (rmt:teststep-set-status! run-id test-id teststep-name state-in status-in comment logfile)
(let* ((valid-values (configf:lookup *configdat* "validvalues" "state"))
(state (items:check-valid-items valid-values "state" state-in))
(status (items:check-valid-items valid-values "status" status-in)))
(if (or (not state)(not status))
(debug:print 3 *default-log-port* "WARNING: Invalid " (if status "status" "state")
" value \"" (if status state-in status-in) "\", update your validvalues section in megatest.config"))
(rmt:send-receive 'teststep-set-status! run-id (list run-id test-id teststep-name state-in status-in comment logfile))))
(define (rmt:delete-steps-for-test! run-id test-id)
(rmt:send-receive 'delete-steps-for-test! run-id (list run-id test-id)))
(define (rmt:get-steps-for-test run-id test-id)
(rmt:send-receive 'get-steps-for-test run-id (list run-id test-id)))
(define (rmt:get-steps-info-by-id test-step-id)
(rmt:send-receive 'get-steps-info-by-id #f (list test-step-id)))
;;======================================================================
;; T E S T D A T A
;;======================================================================
(define (rmt:read-test-data run-id test-id categorypatt #!key (work-area #f))
(rmt:send-receive 'read-test-data run-id (list run-id test-id categorypatt)))
(define (rmt:read-test-data-varpatt run-id test-id categorypatt varpatt #!key (work-area #f))
(rmt:send-receive 'read-test-data-varpatt run-id (list run-id test-id categorypatt varpatt)))
(define (rmt:get-data-info-by-id test-data-id)
(rmt:send-receive 'get-data-info-by-id #f (list test-data-id)))
(define (rmt:testmeta-add-record testname)
(rmt:send-receive 'testmeta-add-record #f (list testname)))
(define (rmt:testmeta-get-record testname)
(rmt:send-receive 'testmeta-get-record #f (list testname)))
(define (rmt:testmeta-update-field test-name fld val)
(rmt:send-receive 'testmeta-update-field #f (list test-name fld val)))
(define (rmt:test-data-rollup run-id test-id status)
(rmt:send-receive 'test-data-rollup run-id (list run-id test-id status)))
(define (rmt:csv->test-data run-id test-id csvdata)
(rmt:send-receive 'csv->test-data run-id (list run-id test-id csvdata)))
;;======================================================================
;; T A S K S
;;======================================================================
(define (rmt:tasks-find-task-queue-records target run-name test-patt state-patt action-patt)
(rmt:send-receive 'find-task-queue-records #f (list target run-name test-patt state-patt action-patt)))
(define (rmt:tasks-add action owner target runname testpatt params)
(rmt:send-receive 'tasks-add #f (list action owner target runname testpatt params)))
(define (rmt:tasks-set-state-given-param-key param-key new-state)
(rmt:send-receive 'tasks-set-state-given-param-key #f (list param-key new-state)))
(define (rmt:tasks-get-last target runname)
(rmt:send-receive 'tasks-get-last #f (list target runname)))
;;======================================================================
;; N O S Y N C D B
;;======================================================================
(define (rmt:no-sync-set var val)
(rmt:send-receive 'no-sync-set #f `(,var ,val)))
(define (rmt:no-sync-get/default var default)
(rmt:send-receive 'no-sync-get/default #f `(,var ,default)))
(define (rmt:no-sync-del! var)
(rmt:send-receive 'no-sync-del! #f `(,var)))
(define (rmt:no-sync-get-lock keyname)
(rmt:send-receive 'no-sync-get-lock #f `(,keyname)))
;;======================================================================
;; A R C H I V E S
;;======================================================================
(define (rmt:archive-get-allocations testname itempath dneeded)
(rmt:send-receive 'archive-get-allocations #f (list testname itempath dneeded)))
(define (rmt:archive-register-block-name bdisk-id archive-path)
(rmt:send-receive 'archive-register-block-name #f (list bdisk-id archive-path)))
(define (rmt:archive-allocate-testsuite/area-to-block block-id testsuite-name areakey)
(rmt:send-receive 'archive-allocate-test-to-block #f (list block-id testsuite-name areakey)))
(define (rmt:archive-register-disk bdisk-name bdisk-path df)
(rmt:send-receive 'archive-register-disk #f (list bdisk-name bdisk-path df)))
(define (rmt:test-set-archive-block-id run-id test-id archive-block-id)
(rmt:send-receive 'test-set-archive-block-id run-id (list run-id test-id archive-block-id)))
(define (rmt:test-get-archive-block-info archive-block-id)
(rmt:send-receive 'test-get-archive-block-info #f (list archive-block-id)))
;; gets mtpg-run-id and syncs the record if different
;;
(define (tasks:run-id->mtpg-run-id dbh cached-info run-id area-info smallest-last-update-time)
(let* ((runs-ht (hash-table-ref cached-info 'runs))
(runinf (hash-table-ref/default runs-ht run-id #f))
(area-id (vector-ref area-info 0)))
(if runinf
runinf ;; already cached
(let* ((run-dat (rmt:get-run-info run-id)) ;; NOTE: get-run-info returns a vector < row header >
(run-name (rmt:get-run-name-from-id run-id))
(row (db:get-rows run-dat)) ;; yes, this returns a single row
(header (db:get-header run-dat))
(state (db:get-value-by-header row header "state"))
(status (db:get-value-by-header row header "status"))
(owner (db:get-value-by-header row header "owner"))
(event-time (db:get-value-by-header row header "event_time"))
(comment (db:get-value-by-header row header "comment"))
(fail-count (db:get-value-by-header row header "fail_count"))
(pass-count (db:get-value-by-header row header "pass_count"))
(db-contour (db:get-value-by-header row header "contour"))
(contour (if (args:get-arg "-prepend-contour")
(if (and db-contour (not (equal? db-contour "")) (string? db-contour ))
(begin
(debug:print-info 10 *default-log-port* "db-contour" db-contour)
db-contour)
(args:get-arg "-contour"))))
(run-tag (if (args:get-arg "-run-tag")
(args:get-arg "-run-tag")
""))
(last-update (db:get-value-by-header row header "last_update"))
(keytarg (if (or (args:get-arg "-prepend-contour") (args:get-arg "-prefix-target"))
(conc "MT_CONTOUR/MT_AREA/" (string-intersperse (rmt:get-keys) "/")) (string-intersperse (rmt:get-keys) "/"))) ;; e.g. version/iteration/platform
(base-target (rmt:get-target run-id))
(target (if (or (args:get-arg "-prepend-contour") (args:get-arg "-prefix-target"))
(conc (or (args:get-arg "-prefix-target") (conc contour "/" (common:get-area-name) "/")) base-target) base-target)) ;; e.g. v1.63/a3e1/ubuntu
(spec-id (pgdb:get-ttype dbh keytarg))
(publish-time (if (args:get-arg "-cp-eventtime-to-publishtime")
event-time
(current-seconds)))
(new-run-id (if (and run-name base-target) (pgdb:get-run-id dbh spec-id target run-name area-id) #f)))
(if new-run-id
(begin ;; let ((run-record (pgdb:get-run-info dbh new-run-id))
(hash-table-set! runs-ht run-id new-run-id)
;; ensure key fields are up to date
;; if last_update == pgdb_last_update do not update smallest-last-update-time
(let* ((pgdb-last-update (pgdb:get-run-last-update dbh new-run-id))
(smallest-time (hash-table-ref/default smallest-last-update-time "smallest-time" #f)))
(if (and (> last-update pgdb-last-update) (or (not smallest-time) (< last-update smallest-time)))
(hash-table-set! smallest-last-update-time "smallest-time" last-update)))
(pgdb:refresh-run-info
dbh
new-run-id
state status owner event-time comment fail-count pass-count area-id last-update publish-time)
(debug:print-info 4 *default-log-port* "Working on run-id " run-id " pgdb-id " new-run-id )
(if (not (equal? run-tag ""))
(task:add-run-tag dbh new-run-id run-tag))
new-run-id)
(if (or (not state) (equal? state "deleted"))
(begin
(debug:print-info 1 *default-log-port* "Warning: Run with id " run-id " was created after previous sync and deleted before the sync") #f)
(if (handle-exceptions
exn
(begin (print-call-chain)
(print ((condition-property-accessor 'exn 'message) exn))
#f)
(pgdb:insert-run
dbh
spec-id target run-name state status owner event-time comment fail-count pass-count area-id last-update publish-time))
(let* ((smallest-time (hash-table-ref/default smallest-last-update-time "smallest-time" #f)))
(if (or (not smallest-time) (< last-update smallest-time))
(hash-table-set! smallest-last-update-time "smallest-time" last-update))
(tasks:run-id->mtpg-run-id dbh cached-info run-id area-info smallest-last-update-time))
#f)))))))
(define (tasks:sync-test-gen-data dbh cached-info test-data-ids smallest-last-update-time)
(let ((test-ht (hash-table-ref cached-info 'tests))
(data-ht (hash-table-ref cached-info 'data)))
(for-each
(lambda (test-data-id)
(let* ((test-data-info (rmt:get-data-info-by-id test-data-id))
(data-id (db:test-data-get-id test-data-info))
(test-id (db:test-data-get-test_id test-data-info))
(category (db:test-data-get-category test-data-info))
(variable (db:test-data-get-variable test-data-info))
(value (db:test-data-get-value test-data-info))
(expected (db:test-data-get-expected test-data-info))
(tol (db:test-data-get-tol test-data-info))
(units (db:test-data-get-units test-data-info))
(comment (db:test-data-get-comment test-data-info))
(status (db:test-data-get-status test-data-info))
(type (db:test-data-get-type test-data-info))
(last-update (db:test-data-get-last_update test-data-info))
(smallest-time (hash-table-ref/default smallest-last-update-time "smallest-time" #f))
(pgdb-test-id (hash-table-ref/default test-ht test-id #f))
(pgdb-data-id (if pgdb-test-id
(pgdb:get-test-data-id dbh pgdb-test-id category variable)
#f)))
(if data-id
(begin
(if pgdb-test-id
(begin
(if pgdb-data-id
(begin
(debug:print-info 4 *default-log-port* "Updating existing test-data with test-id: " test-id " and data-id " data-id " pgdb test id: " pgdb-test-id " pgdb data id " pgdb-data-id)
(let* ((pgdb-last-update (pgdb:get-test-data-last-update dbh pgdb-data-id)))
(if (and (> last-update pgdb-last-update) (or (not smallest-time) (< last-update smallest-time)))
(hash-table-set! smallest-last-update-time "smallest-time" last-update)))
(pgdb:update-test-data dbh pgdb-data-id pgdb-test-id category variable value expected tol units comment status type last-update))
(begin
(debug:print-info 4 *default-log-port* "Inserting test-data with test-id: " test-id " and data-id " data-id " pgdb test id: " pgdb-test-id)
(if (handle-exceptions
exn
(begin (print-call-chain)
(print ((condition-property-accessor 'exn 'message) exn))
#f)
(pgdb:insert-test-data dbh pgdb-test-id category variable value expected tol units comment status type last-update))
;(tasks:run-id->mtpg-run-id dbh cached-info run-id area-info)
(begin
;(pgdb:insert-test-data dbh pgdb-test-id category variable value expected tol units comment status type )
(if (or (not smallest-time) (< last-update smallest-time))
(hash-table-set! smallest-last-update-time "smallest-time" last-update))
(set! pgdb-data-id (pgdb:get-test-data-id dbh pgdb-test-id category variable)))
#f)))
(hash-table-set! data-ht data-id pgdb-data-id ))
(begin
(debug:print-info 1 *default-log-port* "Error: Test not in pgdb"))))
(debug:print-info 1 *default-log-port* "Error: Could not get test data info for data id " test-data-id )))) ;; this is a wierd senario need to debug
test-data-ids)))
(define (task:get-test-times)
(let* ((runname (if (args:get-arg "-runname")
(args:get-arg "-runname")
#f))
(target (if (args:get-arg "-target")
(args:get-arg "-target")
#f))
(test-times (rmt:get-test-times runname target )))
(if (not runname)
(begin
(print "Error: Missing argument -runname")
(exit)))
(if (string-contains runname "%")
(begin
(print "Error: Invalid runname, '%' not allowed (" runname ") ")
(exit)))
(if (not target)
(begin
(print "Error: Missing argument -target")
(exit)))
(if (string-contains target "%")
(begin
(print "Error: Invalid target, '%' not allowed (" target ") ")
(exit)))
(if (eq? (length test-times) 0)
(begin
(print "Data not found!!")
(exit)))
(if (equal? (args:get-arg "-dumpmode") "json")
(task:print-testtime-as-json test-times)
(if (equal? (args:get-arg "-dumpmode") "csv")
(task:print-testtime test-times ",")
(task:print-testtime test-times " ")))))
(define (tasks:sync-test-steps dbh cached-info test-step-ids smallest-last-update-time)
; (print "Sync Steps " test-step-ids )
(let ((test-ht (hash-table-ref cached-info 'tests))
(step-ht (hash-table-ref cached-info 'steps)))
(for-each
(lambda (test-step-id)
(let* ((test-step-info (rmt:get-steps-info-by-id test-step-id))
(step-id (tdb:step-get-id test-step-info))
(test-id (tdb:step-get-test_id test-step-info))
(stepname (tdb:step-get-stepname test-step-info))
(state (tdb:step-get-state test-step-info))
(status (tdb:step-get-status test-step-info))
(event_time (tdb:step-get-event_time test-step-info))
(comment (tdb:step-get-comment test-step-info))
(logfile (tdb:step-get-logfile test-step-info))
(last-update (tdb:step-get-last_update test-step-info))
(pgdb-test-id (hash-table-ref/default test-ht test-id #f))
(smallest-time (hash-table-ref/default smallest-last-update-time "smallest-time" #f))
(pgdb-step-id (if pgdb-test-id
(pgdb:get-test-step-id dbh pgdb-test-id stepname state)
#f)))
(if step-id
(begin
(if pgdb-test-id
(begin
(if pgdb-step-id
(begin
(debug:print-info 4 *default-log-port* "Updating existing test-step with test-id: " test-id " and step-id " step-id " pgdb test id: " pgdb-test-id " pgdb step id " pgdb-step-id )
(let* ((pgdb-last-update (pgdb:get-test-step-last-update dbh pgdb-step-id)))
(if (and (> last-update pgdb-last-update) (or (not smallest-time) (< last-update smallest-time)))
(hash-table-set! smallest-last-update-time "smallest-time" last-update)))
(pgdb:update-test-step dbh pgdb-step-id pgdb-test-id stepname state status event_time comment logfile last-update))
(begin
(debug:print-info 4 *default-log-port* "Inserting test-step with test-id: " test-id " and step-id " step-id " pgdb test id: " pgdb-test-id)
(if (or (not smallest-time) (< last-update smallest-time))
(hash-table-set! smallest-last-update-time "smallest-time" last-update))
(pgdb:insert-test-step dbh pgdb-test-id stepname state status event_time comment logfile last-update )
(set! pgdb-step-id (pgdb:get-test-step-id dbh pgdb-test-id stepname state))))
(hash-table-set! step-ht step-id pgdb-step-id ))
(debug:print-info 1 *default-log-port* "Error: Test not cashed")))
(debug:print-info 1 *default-log-port* "Error: Could not get test step info for step id " test-step-id )))) ;; this is a wierd senario need to debug
test-step-ids)))
(define (tasks:sync-tests-data dbh cached-info test-ids area-info smallest-last-update-time)
(let ((test-ht (hash-table-ref cached-info 'tests)))
(for-each
(lambda (test-id)
; (print test-id)
(let* ((test-info (rmt:get-test-info-by-id #f test-id))
(run-id (db:test-get-run_id test-info)) ;; look these up in db_records.scm
(test-id (db:test-get-id test-info))
(test-name (db:test-get-testname test-info))
(item-path (db:test-get-item-path test-info))
(state (db:test-get-state test-info))
(status (db:test-get-status test-info))
(host (db:test-get-host test-info))
(pid (db:test-get-process_id test-info))
(cpuload (db:test-get-cpuload test-info))
(diskfree (db:test-get-diskfree test-info))
(uname (db:test-get-uname test-info))
(run-dir (db:test-get-rundir test-info))
(log-file (db:test-get-final_logf test-info))
(run-duration (db:test-get-run_duration test-info))
(comment (db:test-get-comment test-info))
(event-time (db:test-get-event_time test-info))
(archived (db:test-get-archived test-info))
(last-update (db:test-get-last_update test-info))
(pgdb-run-id (tasks:run-id->mtpg-run-id dbh cached-info run-id area-info smallest-last-update-time))
(smallest-time (hash-table-ref/default smallest-last-update-time "smallest-time" #f))
(pgdb-test-id (if pgdb-run-id
(begin
;(print pgdb-run-id)
(pgdb:get-test-id dbh pgdb-run-id test-name item-path))
#f)))
;; "id" "run_id" "testname" "state" "status" "event_time"
;; "host" "cpuload" "diskfree" "uname" "rundir" "item_path"
;; "run_duration" "final_logf" "comment" "shortdir" "attemptnum" "archived"
(if (or (not item-path) (string-null? item-path))
(debug:print-info 0 *default-log-port* "Working on Run id : " run-id "and test name : " test-name))
(if pgdb-run-id
(begin
(if pgdb-test-id ;; have a record
(begin ;; let ((key-name (conc run-id "/" test-name "/" item-path)))
(debug:print-info 4 *default-log-port* "Updating existing test with run-id: " run-id " and test-id: " test-id " pgdb run id: " pgdb-run-id " pgdb-test-id " pgdb-test-id)
(let* ((pgdb-last-update (pgdb:get-test-last-update dbh pgdb-test-id)))
(if (and (> last-update pgdb-last-update) (or (not smallest-time) (< last-update smallest-time))) ;;if last-update is same as pgdb-last-update then it is safe to assume the records are identical and we can use a larger last update time.
(hash-table-set! smallest-last-update-time "smallest-time" last-update)))
(pgdb:update-test dbh pgdb-test-id pgdb-run-id test-name item-path state status host cpuload diskfree uname run-dir log-file run-duration comment event-time archived last-update pid))
(begin
(debug:print-info 4 *default-log-port* "Inserting test with run-id: " run-id " and test-id: " test-id " pgdb run id: " pgdb-run-id)
(pgdb:insert-test dbh pgdb-run-id test-name item-path state status host cpuload diskfree uname run-dir log-file run-duration comment event-time archived last-update pid)
(if (or (not smallest-time) (< last-update smallest-time))
(hash-table-set! smallest-last-update-time "smallest-time" last-update))
(set! pgdb-test-id (pgdb:get-test-id dbh pgdb-run-id test-name item-path))))
(hash-table-set! test-ht test-id pgdb-test-id))
(debug:print-info 1 *default-log-port* "WARNING: Skipping run with run-id:" run-id ". This run was created after privious sync and removed before this sync."))))
test-ids)))
;; get runs changed since last sync
;; (define (tasks:sync-test-data dbh cached-info area-info)
;; (let* ((
(define (tasks:sync-to-postgres configdat dest)
(print "In sync")
(let* ((dbh (pgdb:open configdat dbname: dest))
(area-info (pgdb:get-area-by-path dbh *toppath*))
(cached-info (make-hash-table))
(start (current-seconds))
(test-patt (if (args:get-arg "-testpatt")
(args:get-arg "-testpatt")
"%"))
(target (if (args:get-arg "-target")
(args:get-arg "-target")
#f))
(run-name (if (args:get-arg "-runname")
(args:get-arg "-runname")
#f)))
(if (and target (not run-name))
(begin
(print "Error: Provide runname")
(exit 1)))
(if (and (not target) run-name)
(begin
(print "Error: Provide target")
(exit 1)))
;(print "123")
;(exit 1)
(for-each (lambda (dtype)
(hash-table-set! cached-info dtype (make-hash-table)))
'(runs targets tests steps data))
(hash-table-set! cached-info 'start start) ;; when done we'll set sync times to this
(if area-info
(let* ((last-sync-time (vector-ref area-info 3))
(smallest-last-update-time (make-hash-table))
(changed (if (and target run-name)
(rmt:get-run-record-ids target run-name (rmt:get-keys) test-patt)
(rmt:get-changed-record-ids last-sync-time)))
(run-ids (alist-ref 'runs changed))
(test-ids (alist-ref 'tests changed))
(test-step-ids (alist-ref 'test_steps changed))
(test-data-ids (alist-ref 'test_data changed))
(run-stat-ids (alist-ref 'run_stats changed))
(area-tag (if (args:get-arg "-area-tag")
(args:get-arg "-area-tag")
(if (args:get-arg "-area")
(args:get-arg "-area")
""))))
(if (and (equal? area-tag "") (not (pgdb:is-area-taged dbh (vector-ref area-info 0))))
(set! area-tag *default-area-tag*))
(if (not (equal? area-tag ""))
(task:add-area-tag dbh area-info area-tag))
(if (or (not (null? test-ids)) (not (null? run-ids)))
(begin
(debug:print-info 0 *default-log-port* "syncing runs")
(tasks:sync-run-data dbh cached-info run-ids area-info smallest-last-update-time)
(debug:print-info 0 *default-log-port* "syncing tests")
(tasks:sync-tests-data dbh cached-info test-ids area-info smallest-last-update-time)
(debug:print-info 0 *default-log-port* "syncing test steps")
(tasks:sync-test-steps dbh cached-info test-step-ids smallest-last-update-time)
(debug:print-info 0 *default-log-port* "syncing test data")
(tasks:sync-test-gen-data dbh cached-info test-data-ids smallest-last-update-time)
(print "----------done---------------")))
(let* ((smallest-time (hash-table-ref/default smallest-last-update-time "smallest-time" (current-seconds))))
(debug:print-info 0 "smallest-time :" smallest-time " last-sync-time " last-sync-time)
(if (not (and target run-name))
(if (or (and smallest-time (> smallest-time last-sync-time)) (and smallest-time (eq? last-sync-time 0)))
(pgdb:write-sync-time dbh area-info smallest-time))))) ;;this needs to be changed
(if (tasks:set-area dbh configdat)
(tasks:sync-to-postgres configdat dest)
(begin
(debug:print 0 *default-log-port* "ERROR: unable to create an area record")
#f)))))
(define (tasks:sync-run-data dbh cached-info run-ids area-info smallest-last-update-time)
(for-each
(lambda (run-id)
(debug:print-info 4 *default-log-port* "Check if run with " run-id " needs to be synced" )
(tasks:run-id->mtpg-run-id dbh cached-info run-id area-info smallest-last-update-time))
run-ids))
;;======================================================================
;; simple lock. improve and converge on this one.
;;
(define (common:simple-lock keyname)
(rmt:no-sync-get-lock keyname))
(define (common:simple-unlock keyname #!key (force #f))
(rmt:no-sync-del! keyname))
;;======================================================================
;; S T A T E A N D S T A T U S F O R T E S T S
;;======================================================================
;; speed up for common cases with a little logic
(define (mt:test-set-state-status-by-id run-id test-id newstate newstatus newcomment)
(if (not (and run-id test-id))
(begin
(debug:print-error 0 *default-log-port* "bad data handed to mt:test-set-state-status-by-id, run-id=" run-id ", test-id=" test-id ", newstate=" newstate)
(print-call-chain (current-error-port))
#f)
(begin
;; cond
;; ((and newstate newstatus newcomment)
;; (rmt:general-call 'state-status-msg run-id newstate newstatus newcomment test-id))
;; ((and newstate newstatus)
;; (rmt:general-call 'state-status run-id newstate newstatus test-id))
;; (else
;; (if newstate (rmt:general-call 'set-test-state run-id newstate test-id))
;; (if newstatus (rmt:general-call 'set-test-status run-id newstatus test-id))
;; (if newcomment (rmt:general-call 'set-test-comment run-id newcomment test-id))))
(rmt:set-state-status-and-roll-up-items run-id test-id #f newstate newstatus newcomment)
;; (mt:process-triggers run-id test-id newstate newstatus)
#t)))
(define (mt:test-set-state-status-by-id-unless-completed run-id test-id newstate newstatus newcomment)
(let* ((test-vec (rmt:get-testinfo-state-status run-id test-id))
(state (vector-ref test-vec 3)))
(if (equal? state "COMPLETED")
#t
(rmt:set-state-status-and-roll-up-items run-id test-id #f newstate newstatus newcomment))))
(define (mt:test-set-state-status-by-testname run-id test-name item-path new-state new-status new-comment)
;(let ((test-id (rmt:get-test-id run-id test-name item-path)))
(rmt:set-state-status-and-roll-up-items run-id test-name item-path new-state new-status new-comment)
;; (mt:process-triggers run-id test-id new-state new-status)
#t);)
;;(mt:test-set-state-status-by-id run-id test-id new-state new-status new-comment)))
(define (mt:test-set-state-status-by-testname-unless-completed run-id test-name item-path new-state new-status new-comment)
(let ((test-id (rmt:get-test-id run-id test-name item-path)))
(mt:test-set-state-status-by-id-unless-completed run-id test-id new-state new-status new-comment)))
;;======================================================================
;; R U N S
;;======================================================================
;; runs:get-runs-by-patt
;; get runs by list of criteria
;; register a test run with the db
;;
;; Use: (db-get-value-by-header (db:get-header runinfo)(db:get-rows runinfo))
;; to extract info from the structure returned
;;
(define (mt:get-runs-by-patt keys runnamepatt targpatt)
(let loop ((runsdat (rmt:get-runs-by-patt keys runnamepatt targpatt 0 500 #f 0))
(res '())
(offset 0)
(limit 500))
;; (print "runsdat: " runsdat)
(let* ((header (vector-ref runsdat 0))
(runslst (vector-ref runsdat 1))
(full-list (append res runslst))
(have-more (eq? (length runslst) limit)))
;; (debug:print 0 *default-log-port* "header: " header " runslst: " runslst " have-more: " have-more)
(if have-more
(let ((new-offset (+ offset limit))
(next-batch (rmt:get-runs-by-patt keys runnamepatt targpatt offset limit #f 0)))
(debug:print-info 4 *default-log-port* "More than " limit " runs, have " (length full-list) " runs so far.")
(debug:print-info 0 *default-log-port* "next-batch: " next-batch)
(loop next-batch
full-list
new-offset
limit))
(vector header full-list)))))
;;======================================================================
;; T E S T S
;;======================================================================
(define (mt:get-tests-for-run run-id testpatt states status #!key (not-in #t) (sort-by 'event_time) (sort-order "ASC") (qryvals #f)(last-update #f))
(let loop ((testsdat (rmt:get-tests-for-run run-id testpatt states status 0 500 not-in sort-by sort-order qryvals last-update 'normal))
(res '())
(offset 0)
(limit 500))
(let* ((full-list (append res testsdat))
(have-more (eq? (length testsdat) limit)))
(if have-more
(let ((new-offset (+ offset limit)))
(debug:print-info 4 *default-log-port* "More than " limit " tests, have " (length full-list) " tests so far.")
(loop (rmt:get-tests-for-run run-id testpatt states status new-offset limit not-in sort-by sort-order qryvals last-update 'normal)
full-list
new-offset
limit))
full-list))))
(define (mt:lazy-get-prereqs-not-met run-id waitons ref-item-path #!key (mode '(normal))(itemmaps #f) )
(let* ((key (list run-id waitons ref-item-path mode))
(res (hash-table-ref/default *pre-reqs-met-cache* key #f))
(useres (let ((last-time (if (vector? res) (vector-ref res 0) #f)))
(if last-time
(< (current-seconds)(+ last-time 5))
#f))))
(if useres
(let ((result (vector-ref res 1)))
(debug:print 4 *default-log-port* "Using lazy value res: " result)
result)
(let ((newres (rmt:get-prereqs-not-met run-id waitons ref-item-path mode: mode itemmaps: itemmaps)))
(hash-table-set! *pre-reqs-met-cache* key (vector (current-seconds) newres))
newres))))
;;======================================================================
;; from metadat lookup MEGATEST_VERSION
;;
(define (common:get-last-run-version) ;; RADT => How does this work in send-receive function??; assume it is the value saved in some DB
(rmt:get-var #f "MEGATEST_VERSION"))
(define (common:get-last-run-version-number)
(string->number
(substring (common:get-last-run-version) 0 6)))
(define (common:set-last-run-version)
(rmt:set-var #f "MEGATEST_VERSION" (common:version-signature)))
;;======================================================================
;; faux-lock is deprecated. Please use simple-lock below
;;
(define (common:faux-lock keyname #!key (wait-time 8)(allow-lock-steal #t))
(if (rmt:no-sync-get/default keyname #f) ;; do not be tempted to compare to pid. locking is a one-shot action, if already locked for this pid it doesn't actually count
(if (> wait-time 0)
(begin
(thread-sleep! 1)
(if (eq? wait-time 1) ;; only one second left, steal the lock
(begin
(debug:print-info 0 *default-log-port* "stealing lock for " keyname)
(common:faux-unlock keyname force: #t)))
(common:faux-lock keyname wait-time: (- wait-time 1)))
#f)
(begin
(rmt:no-sync-set keyname (conc (current-process-id)))
(equal? (conc (current-process-id)) (conc (rmt:no-sync-get/default keyname #f))))))
(define (common:faux-unlock keyname #!key (force #f))
(if (or force (equal? (conc (current-process-id)) (conc (rmt:no-sync-get/default keyname #f))))
(begin
(if (rmt:no-sync-get/default keyname #f) (rmt:no-sync-del! keyname))
#t)
#f))
;;======================================================================
;; postive number if megatest version > db version
;; negative number if megatest version < db version
(define (common:version-db-delta)
(- megatest-version (common:get-last-run-version-number)))
(define (common:version-changed?)
(not (equal? (common:get-last-run-version)
(common:version-signature))))
(define (common:api-changed?)
(not (equal? (substring (->string megatest-version) 0 4)
(substring (conc (common:get-last-run-version)) 0 4))))
(define (rmt:server-shutdown)
(let ((dbfile (servdat-dbfile *server-info*)))
(debug:print-info 0 *default-log-port* "dbfile is "dbfile)
(if dbfile
(let* ((am-server (args:get-arg "-server"))
(dbfile (args:get-arg "-db"))
(apath *toppath*)
(dbdat (db:get-dbdat *dbstruct-db* apath dbfile))
(db (dbr:dbdat-db dbdat))
(inmem (dbr:dbdat-db dbdat))
)
;; do a final sync here
(debug:print-info 0 *default-log-port* "Doing final sync for "apath" "dbfile" at "(current-seconds))
(db:sync-inmem->disk *dbstruct-db* apath dbfile force-sync: #t)
;; let's finalize here
(debug:print-info 0 *default-log-port* "Finalizing db and inmem")
(sqlite3:finalize! db)
(sqlite3:finalize! inmem)
(debug:print-info 0 *default-log-port* "Finalizing db and inmem complete")
(if am-server
(if (string-match ".*/main.db$" dbfile)
(let ((pkt-file (conc (get-pkts-dir *toppath*)
"/" (servdat-uuid *server-info*)
".pkt")))
(debug:print-info 0 *default-log-port* "removing pkt "pkt-file)
(delete-file* pkt-file)
(debug:print-info 0 *default-log-port* "Releasing lock for "dbfile)
(db:with-lock-db (servdat-dbfile *server-info*)
(lambda (dbh dbfile)
(db:release-lock dbh dbfile))))
(let* ((sdat *server-info*) ;; we have a run-id server
(host (servdat-host sdat))
(port (servdat-port sdat))
(uuid (servdat-uuid sdat)))
(if (not (string-match ".db/main.db" (args:get-arg "-db")))
(let* ((res (rmt:deregister-server *rmt:remote* ;; TODO/BUG: why is this requiring *rmt:remote*?
*toppath*
(servdat-host *server-info*) ;; iface
(servdat-port *server-info*)
(servdat-uuid *server-info*)
(current-process-id)
)))
(debug:print-info 0 *default-log-port* "deregistered-server, res="res)))
(debug:print-info 0 *default-log-port* "deregistering server "host":"port" with uuid "uuid)
)))))))
(define (std-exit-procedure)
;;(common:telemetry-log-close)
(on-exit (lambda () 0))
;;(debug:print-info 13 *default-log-port* "std-exit-procedure called; *time-to-exit*="*time-to-exit*)
(let ((no-hurry (if (bdat-time-to-exit *bdat*) ;; hurry up
#f
(begin
(bdat-time-to-exit-set! *bdat* #t)
#t))))
(debug:print-info 4 *default-log-port* "starting exit process, finalizing databases.")
(if (and no-hurry (debug:debug-mode 18))
(rmt:print-db-stats))
(let ((th1 (make-thread
(lambda () ;; thread for cleaning up, give it five seconds
(let* ((start-time (current-seconds)))
(if (and *server-info*
*unclean-shutdown*)
(begin
(debug:print-info 0 *default-log-port* "Unclean server exit, calling server-shtudown")
(rmt:server-shutdown)))
(debug:print-info 0 *default-log-port* "Shutdown activities completed in "(- (current-seconds) start-time)" seconds"))
;; (if *dbstruct-db* (db:close-all *dbstruct-db*)) ;; one second allocated
#;(if (bdat-task-db *bdat*) ;; TODO: Check that this is correct for task db
(let ((db (cdr (bdat-task-db *bdat*))))
(if (sqlite3:database? db)
(begin
(debug:print-info 0 *default-log-port* "Closing down task db "db)
(sqlite3:interrupt! db)
(sqlite3:finalize! db #t)
(bdat-task-db-set! *bdat* #f)))))
#;(http-client#close-idle-connections!)
(if (not (eq? *default-log-port* (current-error-port)))
(close-output-port *default-log-port*))
(set! *default-log-port* (current-error-port))) "Cleanup db exit thread"))
(th2 (make-thread (lambda ()
(debug:print 4 *default-log-port* "Attempting clean exit. Mode="(if no-hurry "no-hurry" "normal")
" Please be patient and wait a few seconds...")
(if no-hurry
(begin
(thread-sleep! 5)) ;; give the clean up few seconds to do it's stuff
(begin
(thread-sleep! 2)))
(debug:print 4 *default-log-port* " ... done")
)
"clean exit")))
(thread-start! th1)
(thread-start! th2)
(thread-join! th1)
)
)
0)
(define (common:run-sync?)
;; (and (common:on-homehost?)
(args:get-arg "-server"))
;; called in megatest.scm, host-port is string hostname:port
;;
;; NOTE: This is NOT called directly from clients as not all transports support a client running
;; in the same process as the server.
;;
(define (server:ping host port server-id #!key (do-exit #f))
(server-ready? host port server-id))
;;======================================================================
;; http-transportmod.scm contents moved here
;;======================================================================
(define (http-transport:make-server-url hostport)
(if (not hostport)
#f
(conc "http://" (car hostport) ":" (cadr hostport))))
;;======================================================================
;; S E R V E R
;; ======================================================================
(define (http-get-function fnkey)
(hash-table-ref/default *http-functions* fnkey (lambda () "nothing here yet")))
(define (rmt:run hostn)
;; ;; Configurations for server
;; (tcp-buffer-size 2048)
;; (max-connections 2048)
(debug:print 2 *default-log-port* "Attempting to start the server ...")
(let* ((db #f) ;; (open-db)) ;; we don't want the server to be opening and closing the db unnecesarily
(hostname (get-host-name))
(ipaddrstr (let ((ipstr (if (string=? "-" hostn)
;; (string-intersperse (map number->string (u8vector->list (hostname->ip hostname))) ".")
(server:get-best-guess-address hostname)
#f)))
(if ipstr ipstr hostn))) ;; hostname)))
(port (portlogger:open-run-close portlogger:find-port))
(link-tree-path (common:get-linktree))
;; (tmp-area (common:get-db-tmp-area))
#;(start-file (conc tmp-area "/.server-start")))
(debug:print-info 0 *default-log-port* "portlogger recommended port: " port)
(if *server-info*
(begin
(servdat-host-set! *server-info* ipaddrstr)
(servdat-port-set! *server-info* port)
(servdat-status-set! *server-info* 'trying-port)
(servdat-trynum-set! *server-info* (+ (servdat-trynum *server-info*) 1)))
(set! *server-info* (make-servdat host: ipaddrstr port: port)))
(let* ((rep (rmt:try-start-server ipaddrstr port)))
(let loop ((instr (nng-recv rep)))
(let* ((data (string->sexpr instr))
(res (case data
((quit) 'quit)
(else (api:process-request *dbstruct-db* data))))
(resdat (sexpr->string res)))
(if (not (eq? res 'quit))
(begin
(set! *db-last-access* (current-seconds))
(nng-send rep resdat)
(loop (nng-recv rep)))))))
(debug:print-info 0 *default-log-port* "After server, should never see this")
;; server exit stuff here
(let* ((portnum (servdat-port *server-info*)))
(portlogger:open-run-close portlogger:set-port portnum "released")
(rmt:server-shutdown)
;; (bdat-time-to-exit-set! *bdat* #t) ;; tell on-exit to be fast as we've already cleaned up
(portlogger:open-run-close portlogger:set-port port "released") ;; done in rmt:run
;; (debug:print-info 0 *default-log-port* "Max cached queries was " *max-cache-size*)
;; (debug:print-info 0 *default-log-port* "Number of cached writes " *number-of-writes*)
;; (debug:print-info 0 *default-log-port* "Average cached write time "
;; (if (eq? *number-of-writes* 0)
;; "n/a (no writes)"
;; (/ *writes-total-delay*
;; *number-of-writes*))
;; " ms")
;; (debug:print-info 0 *default-log-port* "Number non-cached queries " *number-non-write-queries*)
;; (debug:print-info 0 *default-log-port* "Average non-cached time "
;; (if (eq? *number-non-write-queries* 0)
;; "n/a (no queries)"
;; (/ *total-non-write-delay*
;; *number-non-write-queries*))
;; " ms")
(db:print-current-query-stats)
(debug:print-info 0 *default-log-port* "Server shutdown complete. Exiting")
)))
(define (rmt:try-start-server ipaddrstr portnum)
(if *server-info* ;; update the server info as we might be trying next port
(begin
(servdat-host-set! *server-info* ipaddrstr)
(servdat-port-set! *server-info* portnum)
(servdat-status-set! *server-info* 'trying-port)
(servdat-trynum-set! *server-info*
(+ (servdat-trynum *server-info*) 1)))
(set! *server-info* (make-servdat host: ipaddrstr port: portnum)))
(debug:print-info 0 *default-log-port* "rmt:try-start-server time="
(seconds->time-string (current-seconds))
" ipaddrsstr=" ipaddrstr
" portnum=" portnum)
(if (is-port-in-use portnum)
(begin
(portlogger:open-run-close portlogger:set-failed portnum)
(debug:print 0 *default-log-port* "WARNING: failed to start on portnum: " portnum ", trying next port")
;; (thread-sleep! 0.1)
(rmt:try-start-server ipaddrstr
(portlogger:open-run-close
portlogger:find-port)))
(begin
(if (not *server-info*)
(set! *server-info* (make-servdat
host: ipaddrstr
port: portnum)))
(servdat-status-set! *server-info* 'starting)
(servdat-port-set! *server-info* portnum)
(if (not (servdat-rep *server-info*))
(let ((rep (make-rep-socket)))
(servdat-rep-set! *server-info* rep)
(socket-set! rep 'nng/recvtimeo 2000)))
(let* ((rep (servdat-rep *server-info*)))
(debug:print 0 *default-log-port* "INFO: Trying to start server on " ipaddrstr ":" portnum)
(handle-exceptions
exn
(begin
(print-error-message exn)
(if (< portnum 64000)
(begin
(debug:print 0 *default-log-port* "WARNING: attempt to start server failed. Trying again ...")
(debug:print 0 *default-log-port* " message: " ((condition-property-accessor 'exn 'message) exn))
(debug:print 5 *default-log-port* "exn=" (condition->list exn))
(portlogger:open-run-close portlogger:set-failed portnum)
(debug:print 0 *default-log-port* "WARNING: failed to start on portnum: " portnum ", trying next port")
;; (thread-sleep! 0.1)
(rmt:try-start-server ipaddrstr
(portlogger:open-run-close portlogger:find-port)))
(begin
(print "ERROR: Tried and tried but could not start the server, stopping at port "portnum))))
(nng-listen rep (conc "tcp://*:" portnum))
rep)))))
;;======================================================================
;; S E R V E R U T I L I T I E S
;;======================================================================
;;======================================================================
;; C L I E N T S
;;======================================================================
(define (http-transport:get-time-to-cleanup)
(let ((res #f))
(mutex-lock! *http-mutex*)
(set! res (> (current-seconds) *http-connections-next-cleanup*))
(mutex-unlock! *http-mutex*)
res))
(define (http-transport:inc-requests-count)
(mutex-lock! *http-mutex*)
(set! *http-requests-in-progress* (+ 1 *http-requests-in-progress*))
;; Use this opportunity to slow things down iff there are too many requests in flight
(if (> *http-requests-in-progress* 5)
(begin
(debug:print-info 0 *default-log-port* "Whoa there buddy, ease up...")
(thread-sleep! 1)))
(mutex-unlock! *http-mutex*))
(define (http-transport:dec-requests-count proc)
(mutex-lock! *http-mutex*)
(proc)
(set! *http-requests-in-progress* (- *http-requests-in-progress* 1))
(mutex-unlock! *http-mutex*))
(define (http-transport:dec-requests-count-and-close-all-connections)
(set! *http-requests-in-progress* (- *http-requests-in-progress* 1))
(let loop ((etime (+ (current-seconds) 5))) ;; give up in five seconds
(if (> *http-requests-in-progress* 0)
(if (> etime (current-seconds))
(begin
(thread-sleep! 0.052)
(loop etime))
(debug:print-error 0 *default-log-port*
"requests still in progress after 5 seconds of waiting. I'm going to pass on cleaning up http connections"))
#;(close-idle-connections!)))
(set! *http-connections-next-cleanup* (+ (current-seconds) 10))
(mutex-unlock! *http-mutex*))
(define (http-transport:inc-requests-and-prep-to-close-all-connections)
(mutex-lock! *http-mutex*)
(set! *http-requests-in-progress* (+ 1 *http-requests-in-progress*)))
;; careful closing of connections stored in *runremote*
;;
(define (http-transport:close-connections #!key (area-dat #f))
(debug:print-info 0 *default-log-port* "http-transport:close-connections doesn't do anything now!"))
;; (let* ((runremote (or area-dat *runremote*))
;; (server-dat (if runremote
;; (remote-conndat runremote)
;; #f))) ;; (hash-table-ref/default *runremote* run-id #f)))
;; (if (vector? server-dat)
;; (let ((api-dat (http-transport:server-dat-get-api-uri server-dat)))
;; (handle-exceptions
;; exn
;; (begin
;; (print-call-chain *default-log-port*)
;; (debug:print-error 0 *default-log-port* " closing connection failed with error: " ((condition-property-accessor 'exn 'message) exn) ", exn=" exn))
;; (close-connection! api-dat)
;; ;;(close-idle-connections!)
;; #t))
;; #f)))
;; initialize servdat for client side, setup needed parameters
;; pass in #f as sdat-in to create sdat
;;
#;(define (servdat-init sdat-in iface port uuid)
(let* ((sdat (or sdat-in (make-servdat))))
(assert #f "This is a bad idea.")
(if uuid (servdat-uuid-set! sdat uuid))
(servdat-host-set! sdat iface)
(servdat-port-set! sdat port)
(servdat-api-url-set! sdat (conc "http://" iface ":" port "/api"))
(servdat-api-uri-set! sdat (uri-reference (servdat-api-url sdat)))
(servdat-api-req-set! sdat (make-request method: 'POST
uri: (servdat-api-uri sdat)))
;; set up the http-client parameters
(max-retry-attempts 1)
;; consider all requests indempotent
(retry-request? (lambda (request)
#f))
(determine-proxy (constantly #f))
sdat))
;;======================================================================
;; NEW SERVER METHOD
;;======================================================================
;; only use for main.db - need to re-write some of this :(
;;
(define (get-lock-db sdat dbfile)
(let* ((dbh (db:open-run-db dbfile db:initialize-db))
(res (db:get-iam-server-lock dbh dbfile)))
(sqlite3:finalize! dbh)
res))
(define (register-server pkts-dir pkt-spec host port servkey ipaddr dbpath)
(let* ((pkt-dat `((host . ,host)
(port . ,port)
(servkey . ,servkey)
(pid . ,(current-process-id))
(ipaddr . ,ipaddr)
(dbpath . ,dbpath)))
(uuid (write-alist->pkt
pkts-dir
pkt-dat
pktspec: pkt-spec
ptype: 'server)))
(debug:print 0 *default-log-port* "Server on "host":"port" registered in pkt "uuid)
uuid))
(define (get-pkts-dir #!optional (apath #f))
(let* ((effective-toppath (or *toppath* apath)))
(assert effective-toppath
"ERROR: get-pkts-dir called without *toppath* set. Exiting.")
(let* ((pdir (conc effective-toppath "/.meta/srvpkts")))
(if (file-exists? pdir)
pdir
(begin
(create-directory pdir #t)
pdir)))))
;; given a pkts dir read
;;
(define (get-all-server-pkts pktsdir-in pktspec)
(let* ((pktsdir (if (file-exists? pktsdir-in)
pktsdir-in
(begin
(create-directory pktsdir-in #t)
pktsdir-in)))
(all-pkt-files (glob (conc pktsdir "/*.pkt"))))
(map (lambda (pkt-file)
(read-pkt->alist pkt-file pktspec: pktspec))
all-pkt-files)))
(define (server-address srv-pkt)
(conc (alist-ref 'host srv-pkt) ":"
(alist-ref 'port srv-pkt)))
(define (server-ready? host port key) ;; server-address is host:port
;; (let-values (((i o)(handle-exceptions
;; exn
;; (values #f #f)
;; (tcp-connect host port))))
;; (if (and i o)
(let* ((data (sexpr->string `((cmd . ping)
(key . ,key)
(params . ()))))
(res (open-send-receive-nn (conc host ":" port) data)))
(string->sexpr res)))
;; (let ((res (with-input-from-port i
;; read)))
;; (close-output-port o)
;; (close-input-port i)
;; res))
;; (if (string? res)
;; (string->sexpr res)
;; res)))
;; (begin ;; connection failed
;; (debug:print-info 0 *default-log-port* "Server at "host":"port" is not responding.")
;; #f))))
;; (define (loop-test host port data) ;; server-address is host:port
;; ;; ping the server and ask it
;; ;; if it ready
;; ;; (let* ((sdat (servdat-init #f host port #f)))
;; ;; (http-transport:send-receive sdat "abc" 'ping '())))
;; (let* ((payload (sexpr->string data))
;; (res (with-input-from-request
;; (conc "http://"host":"port"/loop-test")
;; `((data . ,payload))
;; read-string)))
;; (string->sexpr res))
;; #f
;; )
; from the pkts return servers associated with dbpath
;; NOTE: Only one can be alive - have to check on each
;; in the list of pkts returned
;;
(define (get-viable-servers serv-pkts dbpath)
(let loop ((tail serv-pkts)
(res '()))
(if (null? tail)
res ;; NOTE: sort by age so oldest is considered first
(let* ((spkt (car tail)))
(loop (cdr tail)
(if (equal? dbpath (alist-ref 'dbpath spkt))
(cons spkt res)
res))))))
;; from viable servers get one that is alive and ready
;;
(define (get-the-server apath serv-pkts)
(let loop ((tail serv-pkts))
(if (null? tail)
#f
(let* ((spkt (car tail))
(host (alist-ref 'ipaddr spkt))
(port (alist-ref 'port spkt))
(dbpth (alist-ref 'dbpath spkt))
(srvkey (alist-ref 'Z spkt)) ;; (alist-ref 'srvkey spkt))
(addr (server-address spkt)))
(if (server-ready? host port srvkey)
spkt
(loop (cdr tail)))))))
;; am I the "first" in line server? I.e. my D card is smallest
;; use Z card as tie breaker
;;
(define (get-best-candidate serv-pkts dbpath)
(if (null? serv-pkts)
#f
(let loop ((tail serv-pkts)
(best (car serv-pkts)))
(if (null? tail)
best
(let* ((candidate (car tail))
(candidate-bd (string->number (alist-ref 'D candidate)))
(best-bd (string->number (alist-ref 'D best)))
;; bigger number is younger
(candidate-z (alist-ref 'Z candidate))
(best-z (alist-ref 'Z best))
(new-best (cond
((> best-bd candidate-bd) ;; best is younger than candidate
candidate)
((< best-bd candidate-bd) ;; candidate is younger than best
best)
(else
(if (string>=? best-z candidate-z)
best
candidate))))) ;; use Z card as tie breaker
(if (null? tail)
new-best
(loop (cdr tail) new-best)))))))
;;======================================================================
;; END NEW SERVER METHOD
;;======================================================================
;; if .db/main.db check the pkts
;;
(define (http-transport:wait-for-server pkts-dir db-file server-key)
(let* ((sdat *server-info*))
(let loop ((start-time (current-seconds))
(changed #t)
(last-sdat "not this"))
(begin ;; let ((sdat #f))
(thread-sleep! 0.01)
(debug:print-info 0 *default-log-port* "Waiting for server alive signature")
(mutex-lock! *heartbeat-mutex*)
(set! sdat *server-info*)
(mutex-unlock! *heartbeat-mutex*)
(if (and sdat
(not changed)
(> (- (current-seconds) start-time) 2))
(begin
(servdat-status-set! sdat 'iface-stable)
(debug:print-info 0 *default-log-port* "Received server alive signature, now attempting to lock in server")
;; create a server pkt in *toppath*/.meta/srvpkts
;; TODO:
;; 1. change sdat to stuct
;; 2. add uuid to struct
;; 3. update uuid in sdat here
;;
(servdat-uuid-set! sdat
(register-server
pkts-dir *srvpktspec*
(get-host-name)
(servdat-port sdat) server-key
(servdat-host sdat) db-file))
;; (set! *my-signature* (servdat-uuid sdat)) ;; replace with Z, no, stick with proper key
;; now read pkts and see if we are a contender
(let* ((all-pkts (get-all-server-pkts pkts-dir *srvpktspec*))
(viables (get-viable-servers all-pkts db-file))
(best-srv (get-best-candidate viables db-file))
(best-srv-key (if best-srv (alist-ref 'servkey best-srv) #f)))
(debug:print 0 *default-log-port* "best-srv-key: "best-srv-key", server-key: "server-key)
;; am I the best-srv, compare server-keys to know
(if (equal? best-srv-key server-key)
(if (get-lock-db sdat db-file) ;; (db:get-iam-server-lock *dbstruct-db* *toppath* run-id)
(begin
(debug:print 0 *default-log-port* "I'm the server!")
(servdat-dbfile-set! sdat db-file)
(servdat-status-set! sdat 'db-locked))
(begin
(debug:print 0 *default-log-port* "I'm not the server, exiting.")
(bdat-time-to-exit-set! *bdat* #t)
(thread-sleep! 0.2)
(exit)))
(begin
(debug:print 0 *default-log-port*
"Keys do not match "best-srv-key", "server-key", exiting.")
(bdat-time-to-exit-set! *bdat* #t)
(delete-file* (conc (get-pkts-dir *toppath*)
"/" (servdat-uuid *server-info*)
".pkt")) ;; remove immediately instead of waiting for on-exit
(thread-sleep! 0.2)
(exit)))
sdat))
(begin ;; sdat not yet contains server info
(debug:print-info 0 *default-log-port* "Still waiting, last-sdat=" last-sdat)
(sleep 4)
(if (> (- (current-seconds) start-time) 120) ;; been waiting for two minutes
(begin
(debug:print-error 0 *default-log-port* "transport appears to have died, exiting server")
(exit))
(loop start-time
(equal? sdat last-sdat)
sdat))))))))
(define (rmt:register-server remote apath iface port server-key dbname)
(rmt:open-main-connection remote apath) ;; we need a channel to main.db
(rmt:send-receive-real remote apath ;; params: host port servkey pid ipaddr dbpath
(db:run-id->dbname #f)
'register-server `(,iface
,port
,server-key
,(current-process-id)
,iface
,apath
,dbname)))
(define (rmt:get-count-servers remote apath)
(rmt:open-main-connection remote apath) ;; we need a channel to main.db
(rmt:send-receive-real remote apath ;; params: host port servkey pid ipaddr dbpath
(db:run-id->dbname #f)
'get-count-servers `(,apath
)))
(define (rmt:deregister-server remote apath iface port server-key dbname)
(rmt:open-main-connection remote apath) ;; we need a channel to main.db
(rmt:send-receive-real remote apath ;; params: host port servkey pid ipaddr dbpath
(db:run-id->dbname #f)
'deregister-server `(,iface
,port
,server-key
,(current-process-id)
,iface
,apath
,dbname)))
(define (http-transport:wait-for-stable-interface #!optional (num-tries-allowed 100))
;; wait until *server-info* stops changing
(let* ((stime (current-seconds)))
(let loop ((last-host #f)
(last-port #f)
(tries 0))
(let* ((curr-host (and *server-info* (servdat-host *server-info*)))
(curr-port (and *server-info* (servdat-port *server-info*))))
;; first we verify port and interface, update *server-info* in need be.
(cond
((> tries num-tries-allowed)
(debug:print 0 *default-log-port* "http-transport:keep-running, giving up after trying for several minutes.")
(exit 1))
((not *server-info*)
(thread-sleep! 0.25)
(loop curr-host curr-port (+ tries 1)))
((or (not last-host)(not last-port))
(debug:print 0 *default-log-port* "http-transport:keep-running, still no interface, tries="tries)
(thread-sleep! 0.25)
(loop curr-host curr-port (+ tries 1)))
((or (not (equal? last-host curr-host))
(not (equal? last-port curr-port)))
(debug:print-info 0 *default-log-port* "WARNING: interface changed, refreshing iface and port info")
(thread-sleep! 0.25)
(loop curr-host curr-port (+ tries 1)))
((< (- (current-seconds) stime) 1) ;; keep up the looping until at least 3 seconds have passed
(thread-sleep! 0.5)
(loop curr-host curr-port (+ tries 1)))
(else
(rmt:get-signature) ;; sets *my-signature* as side effect
(servdat-status-set! *server-info* 'interface-stable)
(debug:print 0 *default-log-port*
"SERVER STARTED: " curr-host
":" curr-port
" AT " (current-seconds) " server signature: " *my-signature*
" with "(servdat-trynum *server-info*)" port changes")
(flush-output *default-log-port*)
#t))))))
;; run http-transport:keep-running in a parallel thread to monitor that the db is being
;; used and to shutdown after sometime if it is not.
;;
(define (rmt:keep-running dbname)
;; if none running or if > 20 seconds since
;; server last used then start shutdown
;; This thread waits for the server to come alive
(debug:print-info 0 *default-log-port* "Starting the sync-back, keep alive thread in server")
(let* ((server-start-time (current-seconds))
(pkts-dir (get-pkts-dir))
(server-key (rmt:get-signature)) ;; This servers key
(is-main (equal? (args:get-arg "-db") ".db/main.db"))
(last-access 0)
(server-timeout (server:expiration-timeout)))
;; main and run db servers have both got wait logic (could/should merge it)
(if is-main
(http-transport:wait-for-server pkts-dir dbname server-key)
(http-transport:wait-for-stable-interface))
;; this is our forever loop
(let* ((iface (servdat-host *server-info*))
(port (servdat-port *server-info*)))
(let loop ((count 0)
(bad-sync-count 0)
(start-time (current-milliseconds)))
(if (and (not is-main)
(common:low-noise-print 60 "servdat-status"))
(debug:print-info 0 *default-log-port* "servdat-status is " (servdat-status *server-info*)))
;; set up the database handle
(mutex-lock! *heartbeat-mutex*)
(if (not *dbstruct-db*) ;; no db opened yet, open the db and register with main if appropriate
(let ((watchdog (bdat-watchdog *bdat*)))
(debug:print 0 *default-log-port* "SERVER: dbprep")
(db:setup dbname) ;; sets *dbstruct-db* as side effect
(servdat-status-set! *server-info* 'db-opened)
;; IFF I'm not main, call into main and register self
(if (not is-main)
(let ((res (rmt:register-server *rmt:remote*
*toppath* iface port
server-key dbname)))
(if res ;; we are the server
(servdat-status-set! *server-info* 'have-interface-and-db)
(begin
(debug:print 0 *default-log-port* "We are not the server for "dbname", exiting.")
(exit)))))
(debug:print 0 *default-log-port*
"SERVER: running, db "dbname" opened, megatest version: "
(common:get-full-version))
;; start the watchdog
;; is this really needed?
#;(if watchdog
(if (not (member (thread-state watchdog)
'(ready running blocked
sleeping dead)))
(begin
(debug:print-info 0 *default-log-port* "Starting watchdog thread (in state "(thread-state watchdog)")")
(thread-start! watchdog))
(debug:print-info 0 *default-log-port* "Not starting watchdog thread (in state "(thread-state watchdog)")"))
(debug:print 0 *default-log-port* "ERROR: *watchdog* not setup, cannot start it."))
#;(loop (+ count 1) bad-sync-count start-time)
))
(debug:print-info 0 *default-log-port* "syncing "*toppath*" "dbname" at "(current-seconds))
(db:sync-inmem->disk *dbstruct-db* *toppath* dbname force-sync: #t)
(mutex-unlock! *heartbeat-mutex*)
;; when things go wrong we don't want to be doing the various
;; queries too often so we strive to run this stuff only every
;; four seconds or so.
(let* ((sync-time (- (current-milliseconds) start-time))
(rem-time (quotient (- 4000 sync-time) 1000)))
(if (and (<= rem-time 4)
(> rem-time 0))
(thread-sleep! rem-time)))
(if (< count 1) ;; 3x3 = 9 secs aprox
(loop (+ count 1) bad-sync-count (current-milliseconds)))
;; Transfer *db-last-access* to last-access to use in checking that we are still alive
(set! last-access *db-last-access*)
(if (common:low-noise-print 60 "dbstats")
(begin
(debug:print 0 *default-log-port* "Server stats:")
(db:print-current-query-stats)))
(let* ((hrs-since-start (/ (- (current-seconds) server-start-time) 3600)))
(cond
((and *server-run*
(> (+ last-access server-timeout)
(current-seconds))
(if is-main
(> (rmt:get-count-servers *rmt:remote* *toppath*) 1)
#t))
(if (common:low-noise-print 120 "server continuing")
(debug:print-info 0 *default-log-port* "Server continuing, seconds since last db access: " (- (current-seconds) last-access)))
(loop 0 bad-sync-count (current-milliseconds)))
(else
(set! *unclean-shutdown* #f)
(debug:print-info 0 *default-log-port* "Server timed out. seconds since last db access: " (- (current-seconds) last-access))
(debug:print-info 0 *default-log-port* "Starting to shutdown the server. pid="(current-process-id))
(rmt:server-shutdown)
(portlogger:open-run-close portlogger:set-port port "released")
(exit)
#;(debug:print-info 0 *default-log-port* "Sending 'quit to server, received: "
(open-send-receive-nn (conc iface":"port) ;; do this here and not in server-shutdown
(sexpr->string 'quit)))
)))))))
;; Call this to start the actual server
;;
;; all routes though here end in exit ...
;;
;; This is the point at which servers are started
;;
(define (rmt:server-launch dbname)
(debug:print-info 0 *default-log-port* "Entered rmt:server-launch")
(let* ((th2 (make-thread (lambda ()
(debug:print-info 0 *default-log-port* "Server run thread started")
(rmt:run (if (args:get-arg "-server")
(args:get-arg "-server")
"-")
)) "Server run"))
(th3 (make-thread (lambda ()
(debug:print-info 0 *default-log-port* "Server monitor thread started")
(rmt:keep-running dbname)
"Keep running"))))
(thread-start! th2)
(thread-sleep! 0.252) ;; give the server time to settle before starting the keep-running monitor.
(thread-start! th3)
(set! *didsomething* #t)
(thread-join! th2)
(thread-join! th3)
;; (exit))
)
#f
)
;; Generate a unique signature for this process, used at both client and
;; server side
(define (rmt:mk-signature)
(message-digest-string (md5-primitive)
(with-output-to-string
(lambda ()
(write (list (current-directory)
(current-process-id)
(argv)))))))
(define (rmt:get-signature)
(if *my-signature* *my-signature*
(let ((sig (rmt:mk-signature)))
(set! *my-signature* sig)
*my-signature*)))
;;======================================================================
;; Nanomsg transport
;;======================================================================
(define (is-port-in-use port-num)
(let* ((ret #f))
(let-values (((inp oup pid)
(process "netstat" (list "-tulpn" ))))
(let loop ((inl (read-line inp)))
(if (not (eof-object? inl))
(begin
(if (string-search (regexp (conc ":" port-num)) inl)
(begin
;(print "Output: " inl)
(set! ret #t))
(loop (read-line inp)))))))
ret))
;;start a server, returns the connection
;;
(define (start-nn-server portnum )
(let ((rep (make-rep-socket))) ;; (nn-socket 'rep)))
(socket-set! rep 'nng/recvtimeo 2000)
(handle-exceptions
exn
(let ((emsg ((condition-property-accessor 'exn 'message) exn)))
(print "ERROR: Failed to start server \"" emsg "\"")
(exit 1))
(nng-dial #;nn-bind rep (conc "tcp://*:" portnum)))
rep))
;; open connection to server, send message, close connection
;;
(define (open-send-close-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds
(let ((req (make-req-socket 'req))
(uri (conc "tcp://" host-port))
(res #f)
;; (contacts (alist-ref 'contact attrib))
;; (mode (alist-ref 'mode attrib))
)
(socket-set! req 'nng/recvtimeo 2000)
(handle-exceptions
exn
(let ((emsg ((condition-property-accessor 'exn 'message) exn)))
;; Send notification
(debug:print 0 *default-log-port* "ERROR: Failed to connect / send to " uri " message was \"" emsg "\"" )
#f)
(nng-dial req uri)
;; (print "Connected to the server " )
(nng-send req msg)
;; (print "Request Sent")
(let* ((th1 (make-thread (lambda ()
(let ((resp (nng-recv req)))
(nng-close! req)
(set! res (if (equal? resp "ok")
#t
#f))))
"recv thread"))
(th2 (make-thread (lambda ()
(thread-sleep! timeout)
(thread-terminate! th1))
"timer thread")))
(thread-start! th1)
(thread-start! th2)
(thread-join! th1)
res))))
(define (open-send-receive-nn host-port msg #!key (timeout 3) ) ;; default timeout is 3 seconds
(let ((req (make-req-socket))
(uri (conc "tcp://" host-port))
(res #f)
;; (contacts (alist-ref 'contact attrib))
;; (mode (alist-ref 'mode attrib))
)
(handle-exceptions
exn
(let ((emsg ((condition-property-accessor 'exn 'message) exn)))
;; Send notification
(debug:print 0 *default-log-port* "ERROR: Failed to connect / send to " uri " message was \"" emsg "\", exn=" exn)
#f)
(nng-dial req uri)
;; (print "Connected to the server " )
(nng-send req msg)
;; (print "Request Sent")
;; receive code here
;;(print (nn-recv req))
(let* ((th1 (make-thread (lambda ()
(let ((resp (nng-recv req)))
(nng-close! req)
(print resp)
(set! res resp)))
"recv thread"))
(th2 (make-thread (lambda ()
(thread-sleep! timeout)
(thread-terminate! th1))
"timer thread")))
(thread-start! th1)
(thread-start! th2)
(thread-join! th1)
res))))
;;======================================================================
;; S E R V E R U T I L I T I E S
;;======================================================================
;; run ping in separate process, safest way in some cases
;;
#;(define (server:ping-server ifaceport)
(with-input-from-pipe
(conc (common:get-megatest-exe) " -ping " ifaceport)
(lambda ()
(let loop ((inl (read-line))
(res "NOREPLY"))
(if (eof-object? inl)
(case (string->symbol res)
((NOREPLY) #f)
((LOGIN_OK) #t)
(else #f))
(loop (read-line) inl))))))
;; NOT USED (well, ok, reference in rpc-transport but otherwise not used).
;;
#;(define (server:login toppath)
(lambda (toppath)
(set! *db-last-access* (current-seconds)) ;; might not be needed.
(if (equal? *toppath* toppath)
#t
#f)))
;; (define server:sync-lock-token "SERVER_SYNC_LOCK")
;; (define (server:release-sync-lock)
;; (db:no-sync-del! *no-sync-db* server:sync-lock-token))
;; (define (server:have-sync-lock?)
;; (let* ((have-lock-pair (db:no-sync-get-lock *no-sync-db* server:sync-lock-token))
;; (have-lock? (car have-lock-pair))
;; (lock-time (cdr have-lock-pair))
;; (lock-age (- (current-seconds) lock-time)))
;; (cond
;; (have-lock? #t)
;; ((>lock-age
;; (* 3 (configf:lookup-number *configdat* "server" "minimum-intersync-delay" default: 180)))
;; (server:release-sync-lock)
;; (server:have-sync-lock?))
;; (else #f))))
)
;;======================================================================
;; A T T I C
;;======================================================================
;; (handle-directory spiffy-directory-listing)
;; #;(handle-exception (lambda (exn chain)
;; (signal (make-composite-condition
;; (make-property-condition
;; 'server
;; 'message "server error")))))
;;
;; ;; Setup the web server and a /ctrl interface
;; ;;
;; (vhost-map `(((* any) . ,(lambda (continue)
;; ;; open the db on the first call
;; ;; This is were we set up the database connections
;; (let* (($ (request-vars source: 'both))
;; ;; (dat ($ 'dat))
;; (res #f))
;; (cond
;; ((equal? (uri-path (request-uri (current-request)))
;; '(/ "api"))
;; (debug:print 0 *default-log-port* "In api request $=" $)
;; (send-response ;; the $ is the request vars proc
;; body: (http-handle-api *dbstruct-db* $)
;; headers: '((content-type text/plain)))
;; (set! *db-last-access* (current-seconds)))
;; ((equal? (uri-path (request-uri (current-request)))
;; '(/ "ping"))
;; (send-response body: (conc *toppath*"/"(args:get-arg "-db"))
;; headers: '((content-type text/plain))))
;; ((equal? (uri-path (request-uri (current-request)))
;; '(/ "loop-test"))
;; (send-response body: (alist-ref 'data ($))
;; headers: '((content-type text/plain))))
;; ((equal? (uri-path (request-uri (current-request)))
;; '(/ ""))
;; (send-response body: ((http-get-function 'http-transport:main-page))))
;; ((equal? (uri-path (request-uri (current-request)))
;; '(/ "json_api"))
;; (send-response body: ((http-get-function 'http-transport:main-page))))
;; ((equal? (uri-path (request-uri (current-request)))
;; '(/ "runs"))
;; (send-response body: ((http-get-function 'http-transport:main-page))))
;; ((equal? (uri-path (request-uri (current-request)))
;; '(/ any))
;; (send-response body: "hey there!\n"
;; headers: '((content-type text/plain))))
;; ((equal? (uri-path (request-uri (current-request)))
;; '(/ "hey"))
;; (send-response body: "hey there!\n"
;; headers: '((content-type text/plain))))
;; ((equal? (uri-path (request-uri (current-request)))
;; '(/ "jquery3.1.0.js"))
;; (send-response body: ((http-get-function 'http-transport:show-jquery))
;; headers: '((content-type application/javascript))))
;; ((equal? (uri-path (request-uri (current-request)))
;; '(/ "test_log"))
;; (send-response body: ((http-get-function 'http-transport:html-test-log) $)
;; headers: '((content-type text/HTML))))
;; ((equal? (uri-path (request-uri (current-request)))
;; '(/ "dashboard"))
;; (send-response body: ((http-get-function 'http-transport:html-dboard) $)
;; headers: '((content-type text/HTML))))
;; (else (continue))))))))