loadavg.rkt

#lang racket

(require job-queue
         racket/async-channel
         threading)

(define ssh-bin (find-executable-path "ssh"))

;; adjust for your level of safety
(define ssh-args
  (list "-T" "-oStrictHostKeyChecking=no"))

(define (ssh hostname command . args)
  (define remote-command (cons command args))
  (define outp (open-output-string))
  (define exit
    (parameterize ([current-input-port (open-input-bytes #"")]
                   [current-output-port outp])
      (apply system* ssh-bin hostname (append ssh-args remote-command))))
  (and exit (get-output-string outp)))

(define (loadavg hostname)
  (~> (or (ssh hostname "cat" "/proc/loadavg")
          (error 'loadavg "error retrieving data for ~a" hostname))
      string-split
      (take 3)
      (map string->number _)))

(define the-job-queue (make-job-queue 10))

(define (poll-one-host hostname out-ch)
  (submit-job! the-job-queue
               (lambda ()
                 (define res (loadavg hostname))
                 (async-channel-put
                   out-ch (list* (current-seconds) hostname res)))))

(define (poll-hosts hosts out-ch)
  (define (loop)
    (for ([host hosts])
      (poll-one-host host out-ch))
    (sleep 5)
    (loop))
  (loop))

(define (collect-data hostnames)
  (define collection-ch (make-async-channel 10))
  (thread
    (lambda () (poll-hosts hostnames collection-ch)))
  (define (get-message)
    (async-channel-get collection-ch))
  (for ([msg (in-producer get-message)])
    (displayln msg)))

(module* main #f (collect-data (current-command-line-arguments)))