path: root/scripts/guix-build-coordinator-queue-builds-from-guix-data-service.in

#!@GUILE@ --no-auto-compile
-*- scheme -*-
-*- geiser-scheme-implementation: guile -*-
!#
;;; Guix Build Coordinator
;;;
;;; Copyright © 2020 Christopher Baines <mail@cbaines.net>
;;;
;;; This file is part of the guix-build-coordinator.
;;;
;;; guix-data-service 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.
;;;
;;; guix-data-service 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 the guix-data-service.  If not, see
;;; <http://www.gnu.org/licenses/>.

(setvbuf (current-output-port) 'line)
(setvbuf (current-error-port) 'line)

(use-modules (srfi srfi-1)
             (srfi srfi-11)
             (srfi srfi-37)
             (ice-9 match)
             (ice-9 threads)
             (ice-9 textual-ports)
             (rnrs bytevectors)
             (oop goops)
             (logging logger)
             (logging port-log)
             (json)
             (web uri)
             (web client)
             (guix-build-coordinator utils)
             (guix-build-coordinator client-communication))

;; This gets populated from a file on startup
(define processed-commits-hash
  (make-hash-table 1024))

(define processed-derivations-hash
  (make-hash-table 102400))

(define* (guix-data-service-request guix-data-service
                                    path
                                    #:optional (query-parameters '()))
  (define uri
    (string->uri (string-append
                  guix-data-service
                  path
                  (if (null? query-parameters)
                      ""
                      (string-append
                       "?"
                       (string-join
                        (map (match-lambda
                               ((key . value)
                                (simple-format #f "~A=~A" key value)))
                             query-parameters)
                        "&"))))))

  (retry-on-error
   (lambda ()
     (let-values (((response body) (http-get uri)))
       (json-string->scm (utf8->string body))))
   #:times 6
   #:delay (+ 15 (random 30))))

(define (all-repository-ids guix-data-service)
  (let ((data (guix-data-service-request guix-data-service
                                         "/repositories.json")))
    (map (lambda (entry)
           (assoc-ref entry "id"))
         (vector->list
          (assoc-ref data "repositories")))))

(define (all-repository-branches guix-data-service repository-id)
  (let ((data (guix-data-service-request guix-data-service
                                         (string-append
                                          "/repository/"
                                          (number->string repository-id)
                                          ".json"))))
    (map (lambda (entry)
           (assoc-ref entry "name"))
         (vector->list
          (assoc-ref data "branches")))))

(define (unseen-revisions guix-data-service repository-id branch)
  (let ((data (guix-data-service-request guix-data-service
                                         (string-append
                                          "/repository/"
                                          (number->string repository-id)
                                          "/branch/"
                                          branch
                                          ".json"))))
    (filter-map (lambda (entry)
                  (let ((commit (assoc-ref entry "commit-hash")))
                    (and (not (hash-ref processed-commits-hash
                                        commit))
                         (assoc-ref entry "data_available")
                         commit)))
                (vector->list
                 (assoc-ref data "revisions")))))

(define (record-revision-as-processed processed-commits-file commit)
  (monitor
   (let ((port (open-file processed-commits-file "a")))
     (simple-format port "~A\n" commit)
     (close-port port))

   (hash-set! processed-commits-hash commit #t)))

(define (channel-instance-derivations-for-commit guix-data-service
                                                 commit
                                                 system)
  (let ((data (guix-data-service-request
               guix-data-service
               (string-append "/revision/" commit "/channel-instances.json"))))
    (filter-map (lambda (entry)
                  (if (string=? system
                                (assoc-ref entry "system"))
                      (assoc-ref entry "derivation")
                      #f))
                (vector->list
                 (assoc-ref data "channel_instances")))))

(define* (package-derivations-for-commit guix-data-service
                                         commit
                                         #:key system target
                                         guix-data-service-build-server-id)
  (let ((data (guix-data-service-request
               guix-data-service
               (string-append "/revision/" commit "/package-derivations.json")
               `((system      . ,system)
                 (target      . ,target)
                 (field       . "(no-additional-fields)")
                 ,@(if guix-data-service-build-server-id
                       `((no_build_from_build_server
                          . ,guix-data-service-build-server-id))
                       '())
                 (all_results . "on")))))
    (map (lambda (entry)
           (assoc-ref entry "derivation"))
         (vector->list
          (assoc-ref data "derivations")))))

(define (record-derivations-as-processed derivations)
  (monitor
   (for-each (lambda (derivation)
               (hash-set! processed-derivations-hash
                          derivation
                          #t))
             derivations)))

(define* (submit-build coordinator guix-data-service derivation
                       #:key (priority 0) (log-prefix "")
                       (tags '()))
  (retry-on-error
   (lambda ()
     (let ((response
            (send-submit-build-request
             coordinator
             derivation
             (list guix-data-service)
             #f
             priority
             #t
             #t
             #t
             tags)))
       (let ((no-build-submitted-response
              (assoc-ref response "no-build-submitted")))
         (if no-build-submitted-response
             (log-msg 'DEBUG log-prefix "skipped: " no-build-submitted-response)
             (log-msg 'DEBUG log-prefix "build submitted as "
                      (assoc-ref response "build-submitted"))))))
   ;; The TTL Guix uses for transient failures fetching substitutes is 10
   ;; minutes, so we need to retry for longer than that
   #:times 30
   #:delay 30))

(define* (submit-builds-for-revision coordinator
                                     guix-data-service
                                     commit
                                     systems-and-targets
                                     priority-for-derivation
                                     #:key (submit-builds-for-channel-instances?
                                            #t)
                                     guix-data-service-build-server-id
                                     branch)
  (log-msg 'INFO "looking at revision " commit)
  (par-for-each
   (match-lambda
     ((system . target)
      (when (string=? target "none")
        (when submit-builds-for-channel-instances?
          (log-msg 'INFO "requesting channel instance derivations for "
                   system " (" commit ")")
          (for-each (lambda (derivation)
                      (submit-build coordinator
                                    guix-data-service
                                    derivation
                                    #:priority
                                    (priority-for-derivation 'channel-instance
                                                             system
                                                             target)
                                    #:log-prefix
                                    (simple-format #f "channel instance (~A): ~A: "
                                                   system
                                                   derivation)
                                    #:tags `(((key . category)
                                              (value . channel-instance))
                                             ((key . revision)
                                              (value . ,commit))
                                             ,@(if branch
                                                   `(((key . branch)
                                                      (value . ,branch)))
                                                   '()))))
                    (channel-instance-derivations-for-commit guix-data-service
                                                             commit
                                                             system))))

      (let ((unprocessed-package-derivations
             ;; Only request derivations in one thread at a time, just
             ;; in cause doing this in parallel could lead to timeouts
             (monitor
              (log-msg 'INFO "requesting package derivations for "
                       system "=>" target " (" commit ")")
              (package-derivations-for-commit guix-data-service
                                              commit
                                              #:system system
                                              #:target target
                                              #:guix-data-service-build-server-id
                                              guix-data-service-build-server-id))))

        (log-msg 'INFO "submitting package builds for "
                 system "=>" target " (" commit ")")
        (for-each (lambda (derivation)
                    (unless (hash-ref processed-derivations-hash derivation)
                      (submit-build coordinator
                                    guix-data-service
                                    derivation
                                    #:priority
                                    (priority-for-derivation 'package
                                                             system
                                                             target)
                                    #:log-prefix
                                    (simple-format #f "package (~A=>~A): ~A: "
                                                   system target derivation)
                                    #:tags `(((key . category)
                                              (value . package))
                                             ((key . revision)
                                              (value . ,commit))
                                             ,@(if branch
                                                   `(((key . branch)
                                                      (value . ,branch)))
                                                   '())))))
                  unprocessed-package-derivations)
        (record-derivations-as-processed unprocessed-package-derivations))))
   systems-and-targets))

(define %options
  (list (option '("system") #t #f
                (lambda (opt name arg result)
                  (alist-cons 'systems-and-targets
                              `((,arg . "none")
                                ,@(or (assq-ref result 'systems-and-targets) '()))
                              (alist-delete 'systems-and-targets result))))
        (option '("system-and-target") #t #f
                (lambda (opt name arg result)
                  (alist-cons 'systems-and-targets
                              (match (string-split arg #\=)
                                ((system target)
                                 `((,system . ,target)
                                   ,@(or (assq-ref result 'systems-and-targets) '()))))
                              (alist-delete 'systems-and-targets result))))
        (option '("fetch-channel-instance-derivations") #t #f
                (lambda (opt name arg result)
                  (alist-cons 'fetch-channel-instance-derivations
                              (string=? arg "yes")
                              (alist-delete 'fetch-channel-instance-derivations
                                            result))))
        (option '("coordinator") #t #f
                (lambda (opt name arg result)
                  (alist-cons 'coordinator
                              arg
                              (alist-delete 'coordinator result))))
        (option '("guix-data-service") #t #f
                (lambda (opt name arg result)
                  (alist-cons 'guix-data-service
                              arg
                              (alist-delete 'guix-data-service result))))
        (option '("guix-data-service-build-server-id") #t #f
                (lambda (opt name arg result)
                  (alist-cons 'guix-data-service-build-server-id
                              arg
                              (alist-delete 'guix-data-service-build-server-id
                                            result))))
        (option '("branch") #t #f
                (lambda (opt name arg result)
                  (alist-cons 'branches
                              (cons arg
                                    (or (assq-ref result 'branches)
                                        '()))
                              (alist-delete 'branches result))))
        (option '("priority") #t #f
                (lambda (opt name arg result)
                  (alist-cons 'priority
                              (string->number arg)
                              (alist-delete 'priority result))))
        (option '("threads") #t #f
                (lambda (opt name arg result)
                  (alist-cons 'threads
                              (string->number arg)
                              (alist-delete 'threads result))))
        (option '("processed-commits-file") #t #f
                (lambda (opt name arg result)
                  (alist-cons 'processed-commits-file
                              arg
                              (alist-delete 'processed-commits-file result))))))

(define %option-defaults
  '((coordinator                        . "http://127.0.0.1:8746")
    (guix-data-service                  . "https://data.guix.gnu.org")
    (processed-commits-file             . "processed-revisions")
    (fetch-channel-instance-derivations . #t)
    (threads                            . 1)))

(define (parse-options options defaults args)
  (args-fold
   args options
   (lambda (opt name arg result)
     (error "unrecognized option" name))
   (lambda (arg result)
     (alist-cons
      'arguments
      (cons arg
            (or (assoc-ref result 'arguments)
                '()))
      (alist-delete 'arguments result)))
   defaults))

(define (main)
  (let* ((opts (parse-options %options
                              %option-defaults
                              (cdr (program-arguments))))
         (systems-and-targets
          (reverse
           (assq-ref opts 'systems-and-targets)))
         (guix-data-service
          (assq-ref opts 'guix-data-service))
         (processed-commits-file
          (assq-ref opts 'processed-commits-file))

         (lgr      (make <logger>))
         (port-log (make <port-log>
                     #:port (current-output-port)
                     #:formatter
                     (lambda (lvl time str)
                       (format #f "~a (~5a): ~a~%"
                               (strftime "%F %H:%M:%S" (localtime time))
                               lvl
                               str)))))

    (define priority-for-derivation
      (or (and=> (assq-ref opts 'priority)
                 (lambda (fixed-priority)
                   (const fixed-priority)))

          (lambda (type system target)
            (cond
             ((eq? type 'channel-instance)
              (if (and (or (string=? system "x86_64-linux")
                           (string=? system "aarch64-linux"))
                       (string=? target "none"))
                  1000
                  ;; i686-linux builds are getting stuck due to memory issues,
                  ;; so use a lower priority
                  (if (string=? system "i686-linux")
                      0
                      800)))
             ((eq? type 'package)
              (cond
               ((and (or (string=? system "x86_64-linux")
                         (string=? system "aarch64-linux"))
                     (string=? target "none"))
                600)
               (else
                (if (or (string=? target "none")
                        (string=? target "i586-pc-gnu"))
                    400
                    0))))
             (else
              0)))))

    (add-handler! lgr port-log)
    (open-log! lgr)
    (set-default-logger! lgr)

    (unless systems-and-targets
      (simple-format (current-error-port)
                     "error: you must specify at least one system to fetch builds for\n")
      (exit 1))

    (let ((arguments (or (assq-ref opts 'arguments) '())))
      (unless (null? arguments)
        (for-each (lambda (commit)
                    (submit-builds-for-revision
                     (assq-ref opts 'coordinator)
                     guix-data-service
                     commit
                     systems-and-targets
                     priority-for-derivation
                     #:guix-data-service-build-server-id
                     (assq-ref opts 'guix-data-service-build-server-id)))
                  arguments)
        (exit 0)))

    (when (file-exists? processed-commits-file)
      (call-with-input-file processed-commits-file
        (lambda (port)
          (let ((commits
                 (string-split (get-string-all port) #\newline)))
            (for-each (lambda (commit)
                        (unless (string-null? commit)
                          (simple-format #t "marking ~A as already processed\n"
                                         commit)
                          (hash-set! processed-commits-hash
                                     commit
                                     #t)))
                      commits)))))

    (let*-values
        (((process-job-with-queue count-jobs count-threads list-jobs)
          (create-work-queue
           (assq-ref opts 'threads)
           (lambda (branch commit)
             (submit-builds-for-revision
              (assq-ref opts 'coordinator)
              guix-data-service
              commit
              systems-and-targets
              priority-for-derivation
              #:submit-builds-for-channel-instances?
              (assq-ref opts 'fetch-channel-instance-derivations)
              #:guix-data-service-build-server-id
              (assq-ref opts 'guix-data-service-build-server-id)
              #:branch branch)
             (record-revision-as-processed processed-commits-file commit)))))

      (while #t
        (for-each
         (lambda (repository-id)
           (for-each
            (lambda (branch)
              (for-each
               (lambda (commit)
                 (process-job-with-queue branch commit))
               (lset-difference
                string=?
                (unseen-revisions guix-data-service
                                  repository-id
                                  branch)
                (map second (list-jobs)))))
            (let ((all-branches (all-repository-branches guix-data-service
                                                         repository-id))
                  (specified-branches
                   (assq-ref opts 'branches)))
              (or specified-branches
                  all-branches))))
         (all-repository-ids guix-data-service))

        (log-msg 'INFO "jobs: " (count-jobs) " threads: " (count-threads))
        (log-msg 'INFO "waiting before checking for new revisions...")
        (sleep 60)))))

(main)