Use methods for the agent messaging

This will allow adding more agent messaging approaches.

1 files changed, 229 insertions, 178 deletions

diff --git a/guix-build-coordinator/agent-messaging/http.scm b/guix-build-coordinator/agent-messaging/http.scm
index 3c9283f..b1cacdb 100644
--- a/guix-build-coordinator/agent-messaging/http.scm
+++ b/guix-build-coordinator/agent-messaging/http.scm
@@ -29,6 +29,7 @@
   #:use-module (ice-9 binary-ports)
   #:use-module (system repl error-handling)
   #:use-module (rnrs bytevectors)
+  #:use-module (oop goops)
   #:use-module (logging logger)
   #:use-module (json)
   #:use-module (web http)
@@ -48,6 +49,8 @@
   #:export (agent-error-from-coordinator?
             agent-error-from-coordinator-details
 
+            make-http-agent-interface
+
             submit-status
             submit-log-file
             submit-build-result
@@ -61,6 +64,19 @@
   agent-error-from-coordinator?
   (details agent-error-from-coordinator-details))
 
+(define-class <http-agent-interface> ()
+  (coordinator-uri #:init-keyword #:coordinator-uri)
+  (agent-uuid #:init-keyword #:agent-uuid)
+  (password #:init-keyword #:password))
+
+(define (make-http-agent-interface coordinator-uri
+                                   agent-uuid
+                                   password)
+  (make <http-agent-interface>
+    #:coordinator-uri coordinator-uri
+    #:agent-uuid agent-uuid
+    #:password password))
+
 (define (coordinator-uri-for-path base-uri-string agent-path)
   (let* ((base-uri (string->uri base-uri-string))
          (scheme (uri-scheme base-uri))
@@ -102,7 +118,7 @@
       #f)))
 
 (define* (coordinator-http-request log
-                                   coordinator-uri agent-uuid password
+                                   interface
                                    path
                                    #:key method body (headers '())
                                    succeed-on-access-denied-retry?)
@@ -111,10 +127,12 @@
      "Basic "
      (base64-encode
       (string->utf8
-       (string-append agent-uuid ":" password)))))
+       (string-append (slot-ref interface 'agent-uuid)
+                      ":"
+                      (slot-ref interface 'password))))))
 
   (define uri
-    (coordinator-uri-for-path coordinator-uri
+    (coordinator-uri-for-path (slot-ref interface 'coordinator-uri)
                               path))
 
   (define first-request-failed? #f)
@@ -173,186 +191,219 @@
                   #:delay 10
                   #:ignore agent-error-from-coordinator?))
 
-(define* (submit-status coordinator-uri agent-uuid password
-                        status
-                        #:key (log default-log))
-  (coordinator-http-request
-   log
-   coordinator-uri agent-uuid password
-   (string-append "/agent/" agent-uuid)
-   #:method 'PUT                        ; TODO Should be PATCH
-   #:body `((status . ,status))))
-
-(define* (submit-output coordinator-uri agent-uuid password
-                        build-id output-name file
-                        #:key (log default-log))
-  (define auth-value
-    (string-append
-     "Basic "
-     (base64-encode
-      (string->utf8
-       (string-append agent-uuid ":" password)))))
+(define-method (submit-status
+                (interface <http-agent-interface>)
+                .
+                args)
+  (apply
+   (lambda* (status #:key (log default-log))
+     (coordinator-http-request
+      log
+      interface
+      (string-append "/agent/" (slot-ref interface 'agent-uuid))
+      #:method 'PUT                     ; TODO Should be PATCH
+      #:body `((status . ,status))))
+   args))
 
-  (define uri
-    (coordinator-uri-for-path
-     coordinator-uri
-     (string-append "/build/" build-id "/output/" output-name)))
-
-  (define path-info
-    (with-store store
-      (query-path-info store file)))
-
-  ;; For small outputs, compress while sending the data, but for bigger store
-  ;; items, do all the compression up front to hopefully reduce the time to
-  ;; send them.
-  (if (< (path-info-nar-size path-info)
-         5000000)                       ; 5MB
-      (retry-on-error
-       (lambda ()
-         (with-request-mutex
+(define-method (submit-output
+                (interface <http-agent-interface>)
+                .
+                args)
+  (apply
+   (lambda* (build-id output-name file #:key (log default-log))
+     (define auth-value
+       (string-append
+        "Basic "
+        (base64-encode
+         (string->utf8
+          (string-append (slot-ref interface 'agent-uuid)
+                         ":"
+                         (slot-ref interface 'password))))))
+
+     (define uri
+       (coordinator-uri-for-path
+        (slot-ref interface 'coordinator-uri)
+        (string-append "/build/" build-id "/output/" output-name)))
+
+     (define path-info
+       (with-store store
+         (query-path-info store file)))
+
+     ;; For small outputs, compress while sending the data, but for bigger
+     ;; store items, do all the compression up front to hopefully reduce the
+     ;; time to send them.
+     (if (< (path-info-nar-size path-info)
+            5000000)                    ; 5MB
+         (retry-on-error
           (lambda ()
-            (call-with-streaming-http-request
-             uri
+            (with-request-mutex
+             (lambda ()
+               (call-with-streaming-http-request
+                uri
+                (lambda (port)
+                  (call-with-lzip-output-port port
+                    (lambda (port)
+                      (write-file file port))
+                    #:level 9))
+                #:headers `((Authorization . ,auth-value))))))
+          #:times 6
+          #:delay 15)
+         (let* ((directory (or (getenv "TMPDIR") "/tmp"))
+                (template  (string-append directory
+                                          "/guix-build-coordinator-file.XXXXXX"))
+                (out       (mkstemp! template)))
+           (log 'INFO "compressing " file " -> " template " prior to sending")
+           (call-with-lzip-output-port out
              (lambda (port)
-               (call-with-lzip-output-port port
-                 (lambda (port)
-                   (write-file file port))
-                 #:level 9))
-             #:headers `((Authorization . ,auth-value))))))
-       #:times 6
-       #:delay 15)
-      (let* ((directory (or (getenv "TMPDIR") "/tmp"))
-             (template  (string-append directory
-                                       "/guix-build-coordinator-file.XXXXXX"))
-             (out       (mkstemp! template)))
-        (log 'INFO "compressing " file " -> " template " prior to sending")
-        (call-with-lzip-output-port out
-          (lambda (port)
-            (write-file file port))
-          #:level 9)
-        (close-port out)
-
-        (log 'INFO "finished compressing " file ", now sending")
-        (retry-on-error
-         (lambda ()
-           (with-request-mutex
+               (write-file file port))
+             #:level 9)
+           (close-port out)
+
+           (log 'INFO "finished compressing " file ", now sending")
+           (retry-on-error
             (lambda ()
-              (call-with-input-file template
-                (lambda (file-port)
-                  (let-values (((response body)
-                                (call-with-streaming-http-request
-                                 uri
-                                 (lambda (port)
-                                   (with-time-logging
-                                    (simple-format #f "sending ~A" file)
-                                    (dump-port file-port port
-                                               #:buffer-size (expt 2 20))))
-                                 #:headers `((Authorization . ,auth-value)))))
-                    (when (>= (response-code response) 400)
-                      (raise-exception
-                       (make-exception-with-message
-                        (coordinator-handle-failed-request log
-                                                           'PUT
-                                                           (uri-path uri)
-                                                           response
-                                                           body))))))))))
-         #:times 12
-         #:delay (random 15))
-
-        (delete-file template))))
-
-(define* (submit-log-file coordinator-uri agent-uuid password
-                          build-id file
-                          #:key (log default-log))
-  (define auth-value
-    (string-append
-     "Basic "
-     (base64-encode
-      (string->utf8
-       (string-append agent-uuid ":" password)))))
+              (with-request-mutex
+               (lambda ()
+                 (call-with-input-file template
+                   (lambda (file-port)
+                     (let-values (((response body)
+                                   (call-with-streaming-http-request
+                                    uri
+                                    (lambda (port)
+                                      (with-time-logging
+                                          (simple-format #f "sending ~A" file)
+                                        (dump-port file-port port
+                                                   #:buffer-size (expt 2 20))))
+                                    #:headers `((Authorization . ,auth-value)))))
+                       (when (>= (response-code response) 400)
+                         (raise-exception
+                          (make-exception-with-message
+                           (coordinator-handle-failed-request log
+                                                              'PUT
+                                                              (uri-path uri)
+                                                              response
+                                                              body))))))))))
+            #:times 12
+            #:delay (random 15))
 
-  (define format
-    (cond
-     ((string-suffix? ".bz2" file) "bzip2")
-     ((string-suffix? ".gz" file)  "gzip")
-     (else
-      (error "unsupported log format for" file))))
+           (delete-file template))))
+   args))
 
-  (define uri
-    (coordinator-uri-for-path
-     coordinator-uri
-     (string-append "/build/" build-id "/log/" format)))
+(define-method (submit-log-file
+                (interface <http-agent-interface>)
+                .
+                args)
+  (apply
+   (lambda* (build-id file #:key (log default-log))
+     (define auth-value
+       (string-append
+        "Basic "
+        (base64-encode
+         (string->utf8
+          (string-append (slot-ref interface 'agent-uuid)
+                         ":"
+                         (slot-ref interface 'password))))))
+
+     (define format
+       (cond
+        ((string-suffix? ".bz2" file) "bzip2")
+        ((string-suffix? ".gz" file)  "gzip")
+        (else
+         (error "unsupported log format for" file))))
 
-  (retry-on-error
-   (lambda ()
-     (with-request-mutex
+     (define uri
+       (coordinator-uri-for-path
+        (slot-ref interface 'coordinator-uri)
+        (string-append "/build/" build-id "/log/" format)))
+
+     (retry-on-error
       (lambda ()
-        (let-values (((response body)
-                      (call-with-streaming-http-request
-                       uri
-                       (lambda (request-port)
-                         (call-with-input-file file
-                           (lambda (file-port)
-                             (dump-port file-port request-port))
-                           #:binary #t))
-                       #:headers `((Authorization . ,auth-value)))))
-          (if (>= (response-code response) 400)
-              (raise-exception
-               (make-exception-with-message
-                (coordinator-handle-failed-request log
-                                                   'PUT
-                                                   (uri-path uri)
-                                                   response
-                                                   body)))
-              (begin
-                (log 'INFO
-                     "successfully uploaded log file ("
-                     (response-code response)
-                     ")")
-                #t))))))
-   #:times 12
-   #:delay (random 15)))
-
-(define* (submit-build-result coordinator-uri agent-uuid password
-                              build-id result
-                              #:key (log default-log))
-  (coordinator-http-request
-   log
-   coordinator-uri agent-uuid password
-   (string-append "/build/" build-id)
-   #:method 'PUT                        ; TODO Should be PATCH
-   #:body result))
-
-(define* (report-build-start coordinator-uri agent-uuid password
-                             build-id
-                             #:key (log default-log))
-  (coordinator-http-request
-   log
-   coordinator-uri agent-uuid password
-   (string-append "/build/" build-id "/report-build-start")
-   #:method 'POST))
-
-(define* (report-setup-failure coordinator-uri agent-uuid password
-                               build-id report
-                               #:key (log default-log))
-  (coordinator-http-request
-   log
-   coordinator-uri agent-uuid password
-   (string-append "/build/" build-id "/report-setup-failure")
-   #:method 'POST
-   #:body report
-   #:succeed-on-access-denied-retry? #t))
-
-(define* (fetch-builds-for-agent coordinator-uri agent-uuid password
-                                 systems
-                                 target-count
-                                 #:key (log default-log))
-  (vector->list
-   (assoc-ref (coordinator-http-request
-               log
-               coordinator-uri agent-uuid password
-               (string-append "/agent/" agent-uuid "/fetch-builds")
-               #:body `((target_count . ,target-count)
-                        (systems      . ,(list->vector systems)))
-               #:method 'POST)
-              "builds")))
+        (with-request-mutex
+         (lambda ()
+           (let-values (((response body)
+                         (call-with-streaming-http-request
+                          uri
+                          (lambda (request-port)
+                            (call-with-input-file file
+                              (lambda (file-port)
+                                (dump-port file-port request-port))
+                              #:binary #t))
+                          #:headers `((Authorization . ,auth-value)))))
+             (if (>= (response-code response) 400)
+                 (raise-exception
+                  (make-exception-with-message
+                   (coordinator-handle-failed-request log
+                                                      'PUT
+                                                      (uri-path uri)
+                                                      response
+                                                      body)))
+                 (begin
+                   (log 'INFO
+                        "successfully uploaded log file ("
+                        (response-code response)
+                        ")")
+                   #t))))))
+      #:times 12
+      #:delay (random 15)))
+   args))
+
+(define-method (submit-build-result
+                (interface <http-agent-interface>)
+                .
+                args)
+  (apply
+   (lambda* (build-id result #:key (log default-log))
+     (coordinator-http-request
+      log
+      interface
+      (string-append "/build/" build-id)
+      #:method 'PUT                     ; TODO Should be PATCH
+      #:body result))
+   args))
+
+(define-method (report-build-start
+                (interface <http-agent-interface>)
+                .
+                args)
+  (apply
+   (lambda* (build-id #:key (log default-log))
+     (coordinator-http-request
+      log
+      interface
+      (string-append "/build/" build-id "/report-build-start")
+      #:method 'POST))
+   args))
+
+(define-method (report-setup-failure
+                (interface <http-agent-interface>)
+                .
+                args)
+  (apply
+   (lambda* (build-id report #:key (log default-log))
+     (coordinator-http-request
+      log
+      interface
+      (string-append "/build/" build-id "/report-setup-failure")
+      #:method 'POST
+      #:body report
+      #:succeed-on-access-denied-retry? #t))
+   args))
+
+(define-method (fetch-builds-for-agent
+                (interface <http-agent-interface>)
+                .
+                args)
+  (apply
+   (lambda* (systems target-count #:key (log default-log))
+     (vector->list
+      (assoc-ref (coordinator-http-request
+                  log
+                  interface
+                  (string-append "/agent/"
+                                 (slot-ref interface 'agent-uuid)
+                                 "/fetch-builds")
+                  #:body `((target_count . ,target-count)
+                           (systems      . ,(list->vector systems)))
+                  #:method 'POST)
+                 "builds")))
+   args))