Megatest

;;======================================================================
;; Database access
;;======================================================================

(require-extension (srfi 18) extras tcp rpc)
(import (prefix rpc rpc:))

(use sqlite3 srfi-1 posix regex regex-case srfi-69 csv-xml s11n zmq)
(import (prefix sqlite3 sqlite3:))

(declare (unit db))
(declare (uses common))
(declare (uses keys))
(declare (uses ods))

		     #f))))
    (if val
	(begin
	  (debug:print-info 11 "db:set-sync, setting pragma synchronous to " val)
	  (sqlite3:execute db (conc "PRAGMA synchronous = '" val "';"))))))

(define (open-db) ;;  (conc *toppath* "/megatest.db") (car *configinfo*)))
  (if (not *toppath*)(setup-for-run))
  (let* ((dbpath    (conc *toppath* "/megatest.db")) ;; fname)
	 (dbexists  (file-exists? dbpath))
	 (db        (sqlite3:open-database dbpath)) ;; (never-give-up-open-db dbpath))
	 (handler   (make-busy-timeout (if (args:get-arg "-override-timeout")
					   (string->number (args:get-arg "-override-timeout"))
					   136000)))) ;; 136000))) ;; 136000 = 2.2 minutes
    (debug:print-info 11 "open-db, dbpath=" dbpath)

    (set! *global-delta* (/ (+ *global-delta* (* (- (current-milliseconds) start-ms)
						 (if throttle throttle 0.01)))
			    2))
    (if (> (abs (- *last-global-delta-printed* *global-delta*)) 0.08) ;; don't print all the time, only if it changes a bit
	(begin
	  (debug:print-info 4 "launch throttle factor=" *global-delta*)
	  (set! *last-global-delta-printed* *global-delta*)))
    (debug:print-info 11 "db:get-var END " var " val=" res)
    res))

(define (db:set-var db var val)
  (debug:print-info 11 "db:set-var START " var " " val)
  (sqlite3:execute db "INSERT OR REPLACE INTO metadat (var,val) VALUES (?,?);" var val)
  (debug:print-info 11 "db:set-var END " var " " val))

(define (db:del-var db var)
  (debug:print-info 11 "db:del-var START " var)
  (sqlite3:execute db "DELETE FROM metadat WHERE var=?;" var)
  (debug:print-info 11 "db:del-var END " var))

;; use a global for some primitive caching, it is just silly to re-read the db 
;; over and over again for the keys since they never change

(define (db:get-keys db)
  (if *db-keys* *db-keys* 
      (let ((res '()))

                            t.logdat     
                            t.run_duratio
                            t.comment    
                            t.event_time 
                            t.fail_count 
                            t.pass_count 
                            t.archived   



                           FROM tests AS t INNER JOIN runs AS r ON t.run_id=r.id WHERE "
		       keystr " AND r.runname LIKE '" runname "' AND item_path LIKE '" itempatt "' AND testname LIKE '"
		       testpatt "' AND t.state LIKE '" statepatt "' AND t.status LIKE '" statuspatt 
		       "'ORDER BY t.event_time ASC;")))
    (debug:print 3 "qrystr: " qrystr)
    (sqlite3:for-each-row 
     (lambda (p)
       (set! res (cons p res)))
     db 
     qrystr)
    res))

;;======================================================================
;; QUEUE UP META, TEST STATUS AND STEPS
;;======================================================================

;; db:updater is run in a thread to write out the cached data periodically
(define (db:updater)
  (debug:print-info 4 "Starting cache processing")
  (let loop ()
    (thread-sleep! 10) ;; move save time around to minimize regular collisions?
    (db:write-cached-data)
    (loop)))

;; cdb:cached-access is called by the server loop to dispatch commands or queue up
;; db accesses
;;
;; params := qry-name cached? val1 val2 val3 ...
(define (cdb:cached-access params)
  (debug:print-info 12 "cdb:cached-access params=" params)
  (if (< (length params) 2)
      "ERROR"
      (let ((qry-name (car params))
	    (cached?  (cadr params))
	    (remparam (list-tail params 2))) 
	(debug:print-info 12 "cdb:cached-access qry-name=" qry-name " params=" params)
	;; Any special calls are dispatched here. 
	;; Remainder are put in the db queue
	(case qry-name
	  ((login) ;; login checks that the megatest path matches
	   (if (null? remparam)
	       #f ;; no path - fail!
	       (let ((calling-path (car remparam)))
		 (if (equal? calling-path *toppath*)
		     #t      ;; path matches - pass! Should vet the caller at this time ...
		     #f))))  ;; else fail to login
	  ((flush)
	   (db:write-cached-data)
	   #t)
	  (else
	   (mutex-lock! *incoming-mutex*)
	   (set! *last-db-access* (current-seconds))
	   (set! *incoming-data* (cons 
				  (vector qry-name
					  (current-milliseconds)
					  params)
				  *incoming-data*))
	   (mutex-unlock! *incoming-mutex*)
	   ;; NOTE: if cached? is #f then this call must be run immediately
	   ;;       but first all calls in the queue are run first in the order
	   ;;       of their time stamp
	   (if (and cached? *cache-on*)
	       (begin
		 (debug:print-info 12 "*cache-on* is " *cache-on* ", skipping cache write")
		 "CACHED")
	       (begin
		 (db:write-cached-data)
		 "WRITTEN")))))))

(define (db:obj->string obj)(with-output-to-string (lambda ()(serialize obj))))
(define (db:string->obj msg)(with-input-from-string msg (lambda ()(deserialize))))

(define (cdb:client-call zmq-socket . params)
  (debug:print-info 11 "cdb:client-call zmq-socket=" zmq-socket " params=" params)
  (let ((zdat (db:obj->string params)) ;; (with-output-to-string (lambda ()(serialize params))))
	(res  #f))
    (print "cdb:client-call before send message")
    (send-message zmq-socket zdat)
    (print "cdb:client-call after send message")
    (set! res (db:string->obj (receive-message zmq-socket zdat)))
    (debug:print-info 11 "zmq-socket " (car params) " res=" res)
    res))
  
(define (cdb:test-set-status-state zmqsocket test-id status state msg)



  (if msg


      (cdb:client-call zmqsocket 'state-status-msg state status msg test-id)



      (cdb:client-call zmqsocket 'state-status state status test-id))) ;; run-id test-name item-path minutes cpuload diskfree tmpfree) 






(define (cdb:test-rollup-test_data-pass-fail zmqsocket test-id)
  (cdb:client-call zmqsocket 'test_data-pf-rollup #t test-id test-id test-id))

(define (cdb:pass-fail-counts zmqsocket test-id fail-count pass-count)
  (cdb:client-call zmqsocket 'pass-fail-counts fail-count pass-count test-id))

(define (cdb:tests-register-test zmqsocket db run-id test-name item-path)

	     (set! *max-cache-size* cache-size)))
       ))
   #f))

(define cdb:flush-queue db:write-cached-data)

(define (db:roll-up-pass-fail-counts db run-id test-name item-path status)
  




  ;; NEEDED!?
  ;; (rdb:flush-queue)






  (if (and (not (equal? item-path ""))
	   (or (equal? status "PASS")
	       (equal? status "WARN")
	       (equal? status "FAIL")
	       (equal? status "WAIVED")
	       (equal? status "RUNNING")))
      (begin

;; (db:extract-ods-file db "outputfile.ods" '(("sysname" "%")("fsname" "%")("datapath" "%")) "%")


;;======================================================================
;; REMOTE DB ACCESS VIA RPC
;;======================================================================








;; (define (rdb:test-set-status-state test-id status state msg)
;;   (if *runremote*
;;       (let ((host (vector-ref *runremote* 0))
;; 	    (port (vector-ref *runremote* 1)))
;; 	(handle-exceptions
;; 	 exn
;; 	 (begin
;; 	   (debug:print 0 "EXCEPTION: rpc call failed?")
;; 	   (debug:print 0 "  " ((condition-property-accessor 'exn 'message) exn))
;; 	   (print-call-chain)
;; 	   (cdb:test-set-status-state test-id status state msg))
;; 	 ((rpc:procedure 'cdb:test-set-status-state host port) test-id status state msg)))
;;       (cdb:test-set-status-state test-id status state msg)))
;; 
;; (define (rdb:test-rollup-test_data-pass-fail test-id)
;;   (if *runremote*
;;       (let ((host (vector-ref *runremote* 0))
;; 	    (port (vector-ref *runremote* 1)))
;; 	((rpc:procedure 'cdb:test-rollup-test_data-pass-fail host port) test-id))
;;       (cdb:test-rollup-test_data-pass-fail test-id)))
;; 
;; (define (rdb:pass-fail-counts test-id fail-count pass-count)
;;   (if *runremote*
;;       (let ((host (vector-ref *runremote* 0))
;; 	    (port (vector-ref *runremote* 1)))
;; 	((rpc:procedure 'cdb:pass-fail-counts host port) test-id fail-count pass-count))
;;       (cdb:pass-fail-counts test-id fail-count pass-count)))
;; 
;; ;; currently forces a flush of the queue
;; (define (rdb:tests-register-test db run-id test-name item-path)
;;   (if *runremote*
;;       (let ((host (vector-ref *runremote* 0))
;; 	    (port (vector-ref *runremote* 1)))
;; 	((rpc:procedure 'cdb:tests-register-test host port) db run-id test-name item-path force-write: #t))
;;       (cdb:tests-register-test db run-id test-name item-path force-write: #t)))
;; 
;; (define (rdb:flush-queue)
;;   (if *runremote*
;;       (let ((host (vector-ref *runremote* 0))
;; 	    (port (vector-ref *runremote* 1)))
;; 	((rpc:procedure 'cdb:flush-queue host port)))
;;       (cdb:flush-queue)))
;; 

;; Copyright 2006-2012, Matthew Welland.
;; 
;;  This program is made available under the GNU GPL version 2.0 or
;;  greater. See the accompanying file COPYING for details.
;; 
;;  This program is distributed WITHOUT ANY WARRANTY; without even the
;;  implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
;;  PURPOSE.

;; (include "common.scm")
;; (include "megatest-version.scm")

(use sqlite3 srfi-1 posix regex regex-case srfi-69 base64 format readline apropos zmq) ;; (srfi 18) extras)
(import (prefix sqlite3 sqlite3:))
(import (prefix base64 base64:))

(declare (uses common))
(declare (uses megatest-version))
(declare (uses margs))
(declare (uses runs))

;; Misc general calls
;;======================================================================

(if (args:get-arg "-env2file")
    (begin
      (save-environment-as-files (args:get-arg "-env2file"))
      (set! *didsomething* #t)))

;;======================================================================
;; Start the server - can be done in conjunction with -runall or -runtests (one day...)
;;   we start the server if not running else start the client thread
;;======================================================================
(if (args:get-arg "-server")
    (let* ((toppath (setup-for-run))
	   (db      (if toppath (open-db) #f)))
      (debug:print-info 0 "Starting the standalone server")
      (if db 
	  (let* ((th2 (make-thread (lambda ()
				     (server:run (args:get-arg "-server")))))
		 (th3 (make-thread (lambda ()
				     (server:keep-running db)))))
	    (thread-start! th3)
	    (thread-start! th2)
	    (thread-join! th3)
	    (set! *didsomething* #t))
	  (debug:print 0 "ERROR: Failed to setup for megatest")))
    ;; not starting server? then start the client
    (if (server:client-setup)
	(debug:print-info 0 "connected as client")
	(begin
	  (debug:print 0 "ERROR: Failed to connect as client")
	  (exit))))

;;======================================================================
;; Remove old run(s)
;;======================================================================

;; since several actions can be specified on the command line the removal
;; is done first

				       (db:step-get-event_time step)))
			     steps)))))
		  tests))))
	   runs)
	  (set! *didsomething* #t)
	  )))


















;;======================================================================
;; full run
;;======================================================================

;; get lock in db for full run for this directory
;; for all tests with deps
;;   walk tree of tests to find head tasks
;;   add head tasks to task queue
;;   add dependant tasks to task queue 
;;   add remaining tasks to task queue
;; for each task in task queue
;;   if have adequate resources
;;     launch task
;;   else
;;     put task in deferred queue
;; if still ok to run tasks
;;   process deferred tasks per above steps

;; run all tests are are Not COMPLETED and PASS or CHECK
(if (args:get-arg "-runall")
    (let ((server-thread #f))
      (if (args:get-arg "-server")
	  (let ((toppath (setup-for-run))
		(db      (open-db)))
	    (if db 
		(let* ((host:port (db:get-var db "SERVER")) ;; this doen't support multiple servers BUG!!!!
		       (th2 (server:start db (args:get-arg "-server")))
		       (th3 (make-thread (lambda ()
					   (server:keep-running db host:port)))))
		  (thread-start! th3)
		  (set! server-thread th3)))))
      (general-run-call 
       "-runall"
       "run all tests"
       (lambda (target runname keys keynames keyvallst)
	 (runs:run-tests target
			 runname
			 (if (args:get-arg "-testpatt")
			     (args:get-arg "-testpatt")
			     "%/%")
			 user
			 args:arg-hash)))
      (if server-thread 
	  (thread-join! server-thread))))

;;======================================================================
;; run one test
;;======================================================================

;; 1. find the config file
;; 2. change to the test directory

;; Copyright 2006-2012, Matthew Welland.
;; 
;;  This program is made available under the GNU GPL version 2.0 or
;;  greater. See the accompanying file COPYING for details.
;; 
;;  This program is distributed WITHOUT ANY WARRANTY; without even the
;;  implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
;;  PURPOSE.

(require-extension (srfi 18) extras tcp rpc s11n)
(import (prefix rpc rpc:))

(use sqlite3 srfi-1 posix regex regex-case srfi-69 hostinfo zmq)
(import (prefix sqlite3 sqlite3:))

(declare (unit server))

(declare (uses common))
(declare (uses db))
(declare (uses tests))

(include "common_records.scm")
(include "db_records.scm")





(define (server:run hostn)
  (debug:print 0 "Attempting to start the server ...")
  (let ((host:port      (open-run-close db:get-var #f "SERVER"))) ;; do whe already have a server running?
    (if host:port 
	(begin
	  (debug:print 0 "ERROR: server already running.")
	  (if (server:client-setup)
	      (begin 
		(debug:print-info 0 "Server is alive, exiting")
		(exit))
	      (begin
		(debug:print-info 0 "Server is dead, removing flag and trying again")
		(open-run-close db:del-var #f "SERVER")
		(server:run hostn))))
	(let* ((zmq-socket     #f)
	       (hostname       (if (string=? "-" hostn)
				   (get-host-name) 
				   hostn))
	       (ipaddrstr      (let ((ipstr (if (string=? "-" hostn)
						(string-intersperse (map number->string (u8vector->list (hostname->ip hostname))) ".")
						#f)))
				 (if ipstr ipstr hostname))))
	  (set! zmq-socket (server:find-free-port-and-open ipaddrstr zmq-socket 5555))
	  (set! *cache-on* #t)
	  
	  ;; what to do when we quit
	  ;;
	  (on-exit (lambda ()
		     (open-run-close db:del-var #f "SERVER")




		     (let loop () 
		       (let ((queue-len 0))
			 (thread-sleep! (random 5))
			 (mutex-lock! *incoming-mutex*)
			 (set! queue-len (length *incoming-data*))
			 (mutex-unlock! *incoming-mutex*)
			 (if (> queue-len 0)
			     (begin
			       (debug:print-info 0 "Queue not flushed, waiting ...")
			       (loop)))))))

	  ;; The heavy lifting
	  ;;
	  (let loop ()
	    (let* ((rawmsg (receive-message zmq-socket))
		   (params (db:string->obj rawmsg)) ;; (with-input-from-string rawmsg (lambda ()(deserialize))))
		   (res    #f))
	      (debug:print-info 12 "server=> received params=" params)
	      (set! res (cdb:cached-access params))
	      (debug:print-info 12 "server=> processed res=" res)
	      (send-message zmq-socket (db:obj->string res))
	      (loop)))))))

;; run server:keep-running in a parallel thread to monitor that the db is being 
;; used and to shutdown after sometime if it is not.
;;
(define (server:keep-running db)
  ;; if none running or if > 20 seconds since 
  ;; server last used then start shutdown
  (let loop ()
    (thread-sleep! 20) ;; no need to do this very often
    (let ((numrunning (db:get-count-tests-running db)))
      (if (or (> numrunning 0)
	      (> (+ *last-db-access* 60)(current-seconds)))
	  (begin
	    (debug:print-info 0 "Server continuing, tests running: " numrunning ", seconds since last db access: " (- (current-seconds) *last-db-access*))
	    (loop))
	  (begin
	    (debug:print-info 0 "Starting to shutdown the server side")
	    ;; need to delete only *my* server entry (future use)
	    (db:del-var db "SERVER")
	    (thread-sleep! 10)
	    (debug:print-info 0 "Max cached queries was " *max-cache-size*)
	    (debug:print-info 0 "Server shutdown complete. Exiting")
	    ;; (exit)))
	    )))))

(define (server:find-free-port-and-open host s port)
  (let ((s (if s s (make-socket 'rep)))
	(p (if (number? port) port 5555)))
    (handle-exceptions
     exn
     (begin
       (debug:print 0 "Failed to bind to port " p ", trying next port")
       (debug:print 0 "   EXCEPTION: " ((condition-property-accessor 'exn 'message) exn))
       (server:find-free-port-and-open host s (+ p 1)))
     (let ((zmq-url (conc "tcp://" host ":" p)))
       (print "Trying to start server on " zmq-url)
       (bind-socket s zmq-url)
       (set! *runremote* #f)
       (debug:print 0 "Server started on " zmq-url)
       (open-run-close db:set-var #f "SERVER" zmq-url)
       s))))

(define (server:client-setup)




  (let* ((hostinfo   (open-run-close db:get-var #f "SERVER"))
	 (zmq-socket (make-socket 'req)))
    (if hostinfo
	(begin
	  (debug:print-info 2 "Setting up to connect to " hostinfo)
	  (handle-exceptions
	   exn
	   (begin
	     (debug:print 0 "ERROR: Failed to open a connection to the server at: " hostinfo)
	     (debug:print 0 "   EXCEPTION: " ((condition-property-accessor 'exn 'message) exn))
	     (debug:print 0 "   perhaps jobs killed with -9? Removing server records")
	     (open-run-close db:del-var #f "SERVER")
	     (exit)
	     #f)
	   (let ((connect-ok #f))
	     (connect-socket zmq-socket hostinfo)
	     (set! connect-ok (cdb:client-call zmq-socket 'login #t *toppath*))
	     (if connect-ok
		 (begin
		   (debug:print-info 2 "Logged in and connected to " hostinfo)
		   (set! *runremote* zmq-socket)
		   #t)
		 (begin
		   (debug:print-info 2 "Failed to login or connect to " hostinfo)
		   (set! *runremote* #f)
		   #f)))))
	(begin
	  (debug:print-info 2 "No server available, attempting to start one...")
	  (system (conc "megatest -server - " (if (args:get-arg "-debug")
						  (conc "-debug " (args:get-arg "-debug"))
						  "")
			" &"))
	  (sleep 5)
	  (server:client-setup)))))

			     #f))
		       #f)))
    (if waived (set! real-status "WAIVED"))
    (debug:print 4 "real-status " real-status ", waived " waived ", status " status)

    ;; update the primary record IF state AND status are defined
    (if (and state status)
	(cdb:test-set-status-state *runremote* test-id real-status state #f))
    
    ;; if status is "AUTO" then call rollup (note, this one modifies data in test
    ;; run area, do not rpc it (yet)
    (if (and test-id state status (equal? status "AUTO")) 
	(db:test-data-rollup #f test-id status))

    ;; add metadata (need to do this way to avoid SQL injection issues)

	# if [ -e fullrun/megatest.db ]; then sqlite3 fullrun/megatest.db "delete from metadat where var='SERVER';";fi
	mkdir -p /tmp/mt_runs /tmp/mt_links
	cd ..;make install
	rm -f fullrun/logging.db
	touch cleanprep

fullprep : cleanprep
	cd fullrun;$(MEGATEST) -server - &
	sleep 5;cd fullrun;$(MEGATEST) -remove-runs :runname $(RUNNAME)% -target %/%/% -testpatt % -itempatt %
	cd fullrun;$(BINPATH)/dboard -rows 15 &

dashboard : cleanprep
	cd fullrun && $(BINPATH)/dashboard -rows 25 &

remove :
	cd fullrun;$(MEGATEST) -remove-runs :runname $(RUN)  -testpatt % -itempatt % :sysname % :fsname % :datapath %

  --disable-unshare       \
  --disable-rename        \
  --disable-schedutils    \
  --disable-libblkid      \
  --disable-wall
   make install

#  --disable-makeinstall-chown \
#  --disable-makeinstall-setuid \

#   --disable-chsh-only-listed
#   --disable-pg-bell       let pg not ring the bell on invalid keys
#   --disable-require-password
#   --disable-use-tty-group do not install wall and write setgid tty
#   --disable-makeinstall-chown
#   --disable-makeinstall-setuid
    fi