Megatest

;;======================================================================
;; 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.
;;======================================================================

(use srfi-1 posix regex-case base64 format dot-locking csv-xml z3 sql-de-lite hostinfo md5 message-digest typed-records directory-utils stack
     matchable)
     matchable pkts (prefix dbi dbi:))
(require-extension regex posix)

(require-extension (srfi 18) extras tcp rpc)

(import (prefix sqlite3 sqlite3:))
(import (prefix base64 base64:))

;; A hash table that can be accessed by #{scheme ...} calls in
;; config files. Allows communicating between confgs
;;
(define *user-hash-data* (make-hash-table))

(define *db-keys* #f)

(define *pkts-info*    (make-hash-table)) ;; store stuff like the last parent here
(define *configinfo*   #f)   ;; raw results from setup, includes toppath and table from megatest.config
(define *runconfigdat* #f)   ;; run configs data
(define *configdat*    #f)   ;; megatest.config data
(define *configstatus* #f)   ;; status of data; 'fulldata : all processing done, #f : no data yet, 'partialdata : partial read done
(define *toppath*      #f)
(define *already-seen-runconfig-info* #f)

	 (mthome-cfgfile (conc *toppath* "/.mtviews.config")))
    (if (file-exists? mthome-cfgfile)
	(read-config mthome-cfgfile view-cfgdat #t))
    ;; we load the home dir file AFTER the MTRAH file so the user can clobber settings when running the dashboard in read-only areas
    (if (file-exists? home-cfgfile)
	(read-config home-cfgfile view-cfgdat #t))
    view-cfgdat))

;;======================================================================
;; Manage pkts, used in servers, tests and likely other contexts so put
;; in common
;;======================================================================

(define common:pkts-spec
  '((default . ((parent    . P)))
    (server  . ((action    . a)
		(pid       . d)
		(ipaddr    . i)
		(port      . p)
		(parent    . P)))
    			  
    (test    . ((cpuuse    . c)
		(diskuse   . d)
		(item-path . i)
		(runname   . r)
		(state     . s)
		(target    . t)
		(status    . u)
		(parent    . P)))))

(define (common:get-pkts-dirs mtconf use-lt)
  (let* ((pktsdirs-str (or (configf:lookup mtconf "setup"  "pktsdirs")
			   (and use-lt
				(conc *toppath* "/lt/.pkts"))))
	 (pktsdirs  (if pktsdirs-str
			(string-split pktsdirs-str " ")
			#f)))
    pktsdirs))

(define (common:save-pkt pktalist-in mtconf use-lt)
  (let* ((parent   (hash-table-ref/default *pkts-info* 'last-parent #f))
	 (pktalist (if parent
		       (cons `(parent . ,parent)
			     pktalist-in)
		       pktalist-in)))
    (let-values (((uuid pkt)
		  (alist->pkt pktalist common:pkts-spec)))
      (hash-table-set! *pkts-info* 'last-parent uuid)
      (let ((pktsdir (or (hash-table-ref/default *pkts-info* 'pkts-dir #f)
			 (let* ((pktsdirs (common:get-pkts-dirs mtconf use-lt))
			       (pktsdir   (car pktsdirs))) ;; assume it is there
			   (hash-table-set! *pkts-info* 'pkts-dir pktsdir)
			   pktsdir))))
	(if (not (file-exists? pktsdir))
	    (create-directory pktsdir #t))
	(with-output-to-file
	    (conc pktsdir "/" uuid ".pkt")
	  (lambda ()
	    (print pkt)))))))
	
(define (common:with-queue-db mtconf proc #!key (use-lt #f)(toppath-in #f))
  (let* ((pktsdirs (common:get-pkts-dirs mtconf use-lt))
	 (pktsdir  (if pktsdirs (car pktsdirs) #f))
	 (toppath  (or (configf:lookup mtconf "scratchdat" "toppath")
		       toppath-in))
	 (pdbpath  (or (configf:lookup mtconf "setup"  "pdbpath") pktsdir)))
    (if (not (and  pktsdir toppath pdbpath))
	(begin
	  (print "ERROR: settings are missing in your megatest.config for area management.")
	  (print "  you need to have pktsdir in the [setup] section."))
	(let* ((pdb  (open-queue-db pdbpath "pkts.db"
				    schema: '("CREATE TABLE groups (id INTEGER PRIMARY KEY,groupname TEXT, CONSTRAINT group_constraint UNIQUE (groupname));"))))
	  (proc pktsdirs pktsdir pdb)
	  (dbi:close pdb)))))

(define (common:load-pkts-to-db mtconf)
  (common:with-queue-db
   mtconf
   (lambda (pktsdirs pktsdir pdb)
     (for-each
      (lambda (pktsdir) ;; look at all
	(if (and (file-exists? pktsdir)
		 (directory? pktsdir)
		 (file-read-access? pktsdir))
	    (let ((pkts (glob (conc pktsdir "/*.pkt"))))
	      (for-each
	       (lambda (pkt)
		 (let* ((uuid    (cadr (string-match ".*/([0-9a-f]+).pkt" pkt)))
			(exists  (lookup-by-uuid pdb uuid #f)))
		   (if (not exists)
		       (let* ((pktdat (string-intersperse
				       (with-input-from-file pkt read-lines)
				       "\n"))
			      (apkt   (pkt->alist pktdat))
			      (ptype  (alist-ref 'T apkt)))
			 (add-to-queue pdb pktdat uuid (or ptype 'cmd) #f 0)
			 (debug:print 4 *default-log-port* "Added " uuid " of type " ptype " to queue"))
		       (debug:print 4 *default-log-port* "pkt: " uuid " exists, skipping...")
		       )))
	       pkts))))
      pktsdirs))))

(define (common:get-pkt-alists pkts)
  (map (lambda (x)
	 (alist-ref 'apkt x)) ;; 'pkta pulls out the alist from the read pkt
       pkts))

;; given list of pkts (alist mode) return list of D cards as Unix epoch, sorted descending
;; also delete duplicates by target i.e. (car pkt)
;;
(define (common:get-pkt-times pkts)
  (delete-duplicates
   (sort 
    (map (lambda (x)
	   `(,(alist-ref 't x) . ,(string->number (alist-ref 'D x))))
	 pkts)
    (lambda (a b)(> (cdr a)(cdr b))))      ;; sort descending
   (lambda (a b)(equal? (car a)(car b))))) ;; remove duplicates by target

                          (set! sdat *server-info*)
                          (mutex-unlock! *heartbeat-mutex*)
                          (if (and sdat
				   (not changed)
				   (> (- (current-seconds) start-time) 2))
			      (begin
				(debug:print-info 0 *default-log-port* "Received server alive signature")
                                (common:save-pkt `((action . alive)
                                                   (T      . server)
                                                   (pid    . ,(current-process-id))
                                                   (ipaddr . ,(car sdat))
                                                   (port   . ,(cadr sdat)))
                                                 *configdat* #t)
				sdat)
                              (begin
				(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")
                                      (common:save-pkt `((action . died)
                                                         (T      . server)
                                                         (pid    . ,(current-process-id))
                                                         (ipaddr . ,(car sdat))
                                                         (port   . ,(cadr sdat))
                                                         (msg    . "Transport died?"))
                                                 *configdat* #t)
				      (exit))
				    (loop start-time
					  (equal? sdat last-sdat)
					  sdat)))))))
         (iface       (car server-info))
         (port        (cadr server-info))
         (last-access 0)

    ;; 		      (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)
    
    (common:save-pkt `((action . exit)
                       (T      . server)
                       (pid    . ,(current-process-id)))
                     *configdat* #t)
    (debug:print-info 0 *default-log-port* "Server shutdown complete. Exiting")
    (exit)))

;; all routes though here end in exit ...
;;
;; start_server? 
;;
(define (http-transport:launch)
  ;; (if (args:get-arg "-daemonize")
  ;;     (begin
  ;; 	(daemon:ize)
  (common:save-pkt `((action . start)
  ;; 	(if *alt-log-file* ;; we should re-connect to this port, I think daemon:ize disrupts it
  ;; 	    (begin
  ;; 	      (current-error-port *alt-log-file*)
		     (T      . server)
		     (pid    . ,(current-process-id)))
  ;; 	      (current-output-port *alt-log-file*)))))
		   *configdat* #t)
  (let* ((th2 (make-thread (lambda ()
			     (debug:print-info 0 *default-log-port* "Server run thread started")
			     (http-transport:run 
			      (if (args:get-arg "-server")
				  (args:get-arg "-server")
				  "-")
			      )) "Server run"))

[fields]
a text
b text
c text
# [fields]
# a text
# b text
# c text

[setup]
pktsdirs /tmp/pkts /some/other/source

[areas]
#         path-to-area   map-target-script(future, optional)
fullrun   path=tests/fullrun

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

;; fake out readline usage of toplevel-command
(define (toplevel-command . a) #f)

(use srfi-1 posix srfi-69 readline ;;  regex regex-case srfi-69 apropos json http-client directory-utils rpc typed-records;; (srfi 18) extras)
     srfi-18 extras format pkts pkts regex regex-case
     srfi-18 extras format pkts regex regex-case
     (prefix dbi dbi:)) ;;  zmq extras)

(declare (uses common))
(declare (uses megatest-version))
(declare (uses margs))
(declare (uses configf))
;; (declare (uses rmt))

(if (or (args:any? "-h" "help" "-help" "--help")
	(member *action* '("-h" "-help" "--help" "help")))
    (begin
      (print help)
      (exit 1)))

;;======================================================================
;; pkts
;;======================================================================

(define (with-queue-db mtconf proc)
  (let* ((pktsdirs (configf:lookup mtconf "setup"  "pktsdirs"))
	 (pktsdir  (if pktsdirs (car (string-split pktsdirs " ")) #f))
	 (toppath  (configf:lookup mtconf "dyndat" "toppath"))
	 (pdbpath  (or (configf:lookup mtconf "setup"  "pdbpath") pktsdir)))
    (if (not (and  pktsdir toppath pdbpath))
	(begin
	  (print "ERROR: settings are missing in your megatest.config for area management.")
	  (print "  you need to have pktsdir in the [setup] section."))
	(let* ((pdb  (open-queue-db pdbpath "pkts.db"
				    schema: '("CREATE TABLE groups (id INTEGER PRIMARY KEY,groupname TEXT, CONSTRAINT group_constraint UNIQUE (groupname));"))))
	  (proc pktsdirs pktsdir pdb)
	  (dbi:close pdb)))))

(define (load-pkts-to-db mtconf)
  (with-queue-db
   mtconf
   (lambda (pktsdirs pktsdir pdb)
     (for-each
      (lambda (pktsdir) ;; look at all
	(if (and (file-exists? pktsdir)
		 (directory? pktsdir)
		 (file-read-access? pktsdir))
	    (let ((pkts (glob (conc pktsdir "/*.pkt"))))
	      (for-each
	       (lambda (pkt)
		 (let* ((uuid    (cadr (string-match ".*/([0-9a-f]+).pkt" pkt)))
			(exists  (lookup-by-uuid pdb uuid #f)))
		   (if (not exists)
		       (let* ((pktdat (string-intersperse
				       (with-input-from-file pkt read-lines)
				       "\n"))
			      (apkt   (pkt->alist pktdat))
			      (ptype  (alist-ref 'T apkt)))
			 (add-to-queue pdb pktdat uuid (or ptype 'cmd) #f 0)
			 (debug:print 4 *default-log-port* "Added " uuid " of type " ptype " to queue"))
		       (debug:print 4 *default-log-port* "pkt: " uuid " exists, skipping...")
		       )))
	       pkts))))
      (string-split pktsdirs)))))

(define (get-pkt-alists pkts)
  (map (lambda (x)
	 (alist-ref 'apkt x)) ;; 'pkta pulls out the alist from the read pkt
       pkts))

;; given list of pkts (alist mode) return list of D cards as Unix epoch, sorted descending
;; also delete duplicates by target i.e. (car pkt)
(define (get-pkt-times pkts)
  (delete-duplicates
   (sort 
    (map (lambda (x)
	   `(,(alist-ref 't x) . ,(string->number (alist-ref 'D x))))
	 pkts)
    (lambda (a b)(> (cdr a)(cdr b))))      ;; sort descending
   (lambda (a b)(equal? (car a)(car b))))) ;; remove duplicates by target

;;======================================================================
;; Runs
;;======================================================================

;; make a runname
;;
(define (make-runname pre post)

	 (mtconfdat (find-and-read-config        ;; NB// sets MT_RUN_AREA_HOME as side effect
		     mtconfig
		     ;; environ-patt: "env-override"
		     given-toppath: start-dir
		     ;; pathenvvar: "MT_RUN_AREA_HOME"
		     ))
	 (mtconf    (if mtconfdat (car mtconfdat) #f)))
    ;; we set some dynamic data in a section called "dyndata"
    ;; we set some dynamic data in a section called "scratchdata"
    (if mtconf
	(begin
	  (configf:section-var-set! mtconf "dyndat" "toppath" start-dir)))
    ;; (print "TOPPATH: " (configf:lookup mtconf "dyndat" "toppath"))
	  (configf:section-var-set! mtconf "scratchdat" "toppath" start-dir)))
    ;; (print "TOPPATH: " (configf:lookup mtconf "scratchdat" "toppath"))
    mtconfdat))


;; NEED TIMESTAMP ON PKTS for efficient loading of packets into db.


;; make a run request pkt from basic data, this seriously needs to be refactored

;; (use trace)(trace create-run-pkt)

;; collect all needed data and create run pkts for contours with changed inputs
;;
(define (generate-run-pkts mtconf toppath)
  (let ((std-runname (conc "sched"  (time->string (seconds->local-time (current-seconds)) "%M%H%d"))))
    (with-queue-db
    (common:with-queue-db
     mtconf
     (lambda (pktsdirs pktsdir pdb)
       (let* ((rgconfdat (find-and-read-config (conc toppath "/runconfigs.config")))
	      (rgconf    (car rgconfdat))
	      (all-areas (map car (configf:get-section mtconf "areas")))
	      (contours  (configf:get-section mtconf "contours"))
	      (torun     (make-hash-table)) ;; target => ( ... info ... )

			;; (val-list   (string-split-fields ";\\s*" val #:infix)) ;; (string-split val)) ;; runname-rule params
			(val-alist  (val->alist val))
			(runname    (make-runname "" ""))
			(runtrans   (alist-ref 'runtrans val-alist))
			
			(runstarts  (find-pkts pdb '(runstart) `((o . ,contour)
								 (t . ,runkey))))
			(rspkts     (get-pkt-alists runstarts))
			(rspkts     (common:get-pkt-alists runstarts))
			;; starttimes is for run start times and is used to know when the last run was launched
			(starttimes (get-pkt-times rspkts)) ;; sort by age (youngest first) and delete duplicates by target
			(starttimes (common:get-pkt-times rspkts)) ;; sort by age (youngest first) and delete duplicates by target
			(last-run (if (null? starttimes) ;; if '() then it has never been run, else get the max
				      0
				      (apply max (map cdr starttimes))))
			;; synctimes is for figuring out the last time a sync was done
			(syncstarts   (find-pkts pdb '(syncstart) '())) ;; no qualifiers, a sync does all tarets etc.
			(sspkts       (get-pkt-alists syncstarts))
			(synctimes    (get-pkt-times  sspkts))
			(sspkts       (common:get-pkt-alists syncstarts))
			(synctimes    (common:get-pkt-times  sspkts))
			(last-sync (if (null? synctimes) ;; if '() then it has never been run, else get the max
				      0
				      (apply max (map cdr synctimes))))
			)

		   (let ((delta (lambda (x)
				  (round (/ (- (current-seconds) x) 60)))))

		     exn
		     #f
		   (create-directory "logs")
		   #t)
		 #t)
	     "logs"
	     "/tmp")))
    (with-queue-db
    (common:with-queue-db
     mtconf
     (lambda (pktsdirs pktsdir pdb)
       (let* ((rgconfdat (find-and-read-config (conc toppath "/runconfigs.config")))
	      (rgconf    (car rgconfdat))
	      (areas     (configf:get-section mtconf "areas"))
	      (contours  (configf:get-section mtconf "contours"))
	      (pkts      (find-pkts pdb '(cmd) '()))

	 ;;  (hash-table-keys adjargs))
	 (let-values (((uuid pkt)
		       (command-line->pkt *action* adjargs #f)))
	   (write-pkt pktsdir uuid pkt))))
      ((dispatch import rungen process)
       (let* ((mtconfdat (simple-setup (args:get-arg "-start-dir")))
	      (mtconf    (car mtconfdat))
	      (toppath   (configf:lookup mtconf "dyndat" "toppath")))
	      (toppath   (configf:lookup mtconf "scratchdat" "toppath")))
	 (case (string->symbol *action*)
	   ((process)  (begin
			 (load-pkts-to-db mtconf)
			 (common:load-pkts-to-db mtconf)
			 (generate-run-pkts mtconf toppath)
			 (load-pkts-to-db mtconf)
			 (common:load-pkts-to-db mtconf)
			 (dispatch-commands mtconf toppath)))
	   ((import)   (load-pkts-to-db mtconf)) ;; import pkts
	   ((import)   (common:load-pkts-to-db mtconf)) ;; import pkts
	   ((rungen)   (generate-run-pkts mtconf toppath))
	   ((dispatch) (dispatch-commands mtconf toppath)))))
      ((db)
       (if (null? remargs)
	   (print "ERROR: missing sub command for db command")
	   (let ((subcmd (car remargs)))
	     (case (string->symbol subcmd)

;; 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 s11n)

(use srfi-1 posix regex regex-case srfi-69 hostinfo md5 message-digest directory-utils posix-extras matchable)
;; (use zmq)
(use srfi-1 posix regex regex-case srfi-69 hostinfo md5 message-digest
     directory-utils posix-extras matchable)

(use spiffy uri-common intarweb http-client spiffy-request-vars)

(declare (unit server))

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

(define (server:make-server-url hostport)
  (if (not hostport)
      #f
      (conc "http://" (car hostport) ":" (cadr hostport))))

(define  *server-loop-heart-beat* (current-seconds))

;;======================================================================
;; P K T S   S T U F F 
;;======================================================================

;; ???

;;======================================================================
;; S E R V E R
;;======================================================================

;; Call this to start the actual server
;;

# this may save a few milliseconds on launching tests
# launchwait no
waivercommentpatt ^WW\d+ [a-z].*
incomplete-timeout 1

# wait 0.5 seconds between launching every process
#
launch-delay 0.5
# launch-delay 0.5
launch-delay 0


# wait for runs to completely complete. yes, anything else is no
run-wait yes

# If set to "default" the old code is used. Otherwise defaults to 200 or uses
# numeric value given.
#