|
18 | 18 | [babashka.fs :as fs] |
19 | 19 | [cheshire.core :as json] |
20 | 20 | [clojure.core.async :as async] |
| 21 | + [clojure.java.io :as io] |
21 | 22 | [clojure.string :as string] |
22 | 23 | [creds] |
23 | 24 | jsonrpc |
|
27 | 28 | schema |
28 | 29 | shutdown) |
29 | 30 | (:import |
| 31 | + [java.io PipedInputStream PipedOutputStream] |
30 | 32 | [java.net UnixDomainSocketAddress] |
31 | 33 | [java.nio ByteBuffer] |
32 | 34 | [java.nio.channels SocketChannel] |
|
100 | 102 | :password pat |
101 | 103 | :serveraddress "https://index.docker.io/v1/"}}))) |
102 | 104 |
|
| 105 | +(defn list-containers [m] |
| 106 | + (curl/get |
| 107 | + "http://localhost/containers/json" |
| 108 | + {:raw-args ["--unix-socket" "/var/run/docker.sock"] |
| 109 | + :throw false})) |
| 110 | + |
103 | 111 | (defn list-images [m] |
104 | 112 | (curl/get |
105 | 113 | "http://localhost/images/json" |
|
153 | 161 | ;; Tty wraps the process in a pseudo terminal |
154 | 162 | ;; StdinOnce closes the stdin after the first client detaches |
155 | 163 | ;; OpenStdin just opens stdin |
156 | | -(defn create-container [{:keys [image entrypoint workdir command host-dir environment thread-id opts mounts volumes ports network_mode secrets] |
157 | | - :or {opts {:Tty true}}}] |
| 164 | +(defn create-container [{:keys [image entrypoint workdir command host-dir |
| 165 | + environment thread-id opts mounts volumes |
| 166 | + ports network_mode secrets labels] |
| 167 | + :or {opts {:Tty true}} |
| 168 | + :as m}] |
158 | 169 | (let [payload (json/generate-string |
159 | 170 | (merge |
160 | 171 | {:Image image} |
|
163 | 174 | {:Env (->> environment |
164 | 175 | (map (fn [[k v]] (format "%s=%s" (name k) v))) |
165 | 176 | (into []))}) |
166 | | - {:Labels (->> secrets |
167 | | - keys |
168 | | - (map (fn [secret-key] (let [s (name secret-key)] [(format "x-secret:%s" s) (string/trim (format "/secret/%s" s))]))) |
169 | | - (into {}))} |
| 177 | + {:Labels (->> (concat |
| 178 | + (->> secrets |
| 179 | + keys |
| 180 | + (map (fn [secret-key] |
| 181 | + (let [s (name secret-key)] |
| 182 | + [(format "x-secret:%s" s) (string/trim (format "/secret/%s" s))])))) |
| 183 | + (seq labels)) |
| 184 | + (into {}))} |
170 | 185 | {:HostConfig |
171 | 186 | (merge |
172 | 187 | {:Binds |
|
185 | 200 | (when command {:Cmd command}))) |
186 | 201 | ascii-payload (String. (.getBytes payload "ASCII"))] |
187 | 202 | (curl/post |
188 | | - "http://localhost/containers/create" |
| 203 | + (format |
| 204 | + "http://localhost/containers/create%s" |
| 205 | + (if (:name m) (format "?name=%s" (:name m)) "")) |
189 | 206 | {:raw-args ["--unix-socket" "/var/run/docker.sock"] |
190 | 207 | :throw false |
191 | 208 | :body ascii-payload |
|
293 | 310 | (def get-archive (comp (status? 200 "container->archive") container->archive)) |
294 | 311 | (def pull (comp (status? 200 "pull-image") pull-image)) |
295 | 312 | (def images (comp ->json list-images)) |
| 313 | +(def containers (comp ->json (status? 200 "list-containers") list-containers)) |
296 | 314 |
|
297 | 315 | (defn add-latest [image] |
298 | 316 | (let [[_ tag] (re-find #".*(:.*)$" image)] |
|
478 | 496 | (.flip ^ByteBuffer buf) |
479 | 497 | (while (.hasRemaining buf) |
480 | 498 | (.write client buf)) |
481 | | - (catch Throwable _ |
| 499 | + (catch Throwable ex |
| 500 | + (logger/error "write-stdin error " ex) |
482 | 501 | client)) |
483 | 502 | client)) |
484 | 503 |
|
|
531 | 550 | c - channel to write multiplexed stdout stderr blocks" |
532 | 551 | [in c] |
533 | 552 | (try |
534 | | - (let [header-buf (ByteBuffer/allocate 8)] |
| 553 | + (let [header-buf (ByteBuffer/allocate 8) |
| 554 | + stdout (PipedOutputStream.) |
| 555 | + stdout-reader (io/reader (PipedInputStream. stdout))] |
| 556 | + (async/go-loop [] |
| 557 | + (when-let [line (.readLine stdout-reader)] |
| 558 | + (async/put! c {:stdout line}) |
| 559 | + (recur))) |
535 | 560 | (loop [offset 0] |
536 | 561 | (let [result (.read ^SocketChannel in header-buf)] |
537 | 562 | (cond |
538 | 563 | ;;;;;;;;;; |
539 | 564 | (= -1 result) |
540 | | - (async/close! c) |
| 565 | + (do |
| 566 | + (.close stdout) |
| 567 | + (async/close! c)) |
541 | 568 |
|
542 | 569 | ;;;;;;;;;; |
543 | 570 | (= 8 (+ offset result)) |
544 | | - (do |
545 | | - (let [size (.getInt (ByteBuffer/wrap (Arrays/copyOfRange ^bytes (.array ^ByteBuffer header-buf) 4 8))) |
546 | | - stream-type (case (int (nth (.array ^ByteBuffer header-buf) 0)) |
547 | | - 0 :stdin |
548 | | - 1 :stdout |
549 | | - 2 :stderr) |
550 | | - buf (ByteBuffer/allocate size)] |
551 | | - (.clear ^ByteBuffer header-buf) |
552 | | - (loop [offset 0] |
553 | | - (let [result (.read ^SocketChannel in buf)] |
554 | | - (cond |
| 571 | + |
| 572 | + (let [size (.getInt (ByteBuffer/wrap (Arrays/copyOfRange ^bytes (.array ^ByteBuffer header-buf) 4 8))) |
| 573 | + stream-type (case (int (nth (.array ^ByteBuffer header-buf) 0)) |
| 574 | + 0 :stdin |
| 575 | + 1 :stdout |
| 576 | + 2 :stderr) |
| 577 | + buf (ByteBuffer/allocate size)] |
| 578 | + (.clear ^ByteBuffer header-buf) |
| 579 | + (loop [offset 0] |
| 580 | + (let [result (.read ^SocketChannel in buf)] |
| 581 | + (cond |
555 | 582 | ;;;;;;;;;; |
556 | | - (= -1 result) |
557 | | - (async/close! c) |
| 583 | + (= -1 result) |
| 584 | + (async/close! c) |
558 | 585 |
|
559 | 586 | ;;;;;;;;;; |
560 | | - (= size (+ offset result)) |
561 | | - (async/put! c {stream-type (String. ^bytes (.array buf))}) |
| 587 | + (= size (+ offset result)) |
| 588 | + (if (= stream-type :stdout) |
| 589 | + (do |
| 590 | + (.write stdout ^bytes (.array buf)) |
| 591 | + (.flush stdout)) |
| 592 | + (async/put! c {stream-type (String. ^bytes (.array buf))})) |
562 | 593 |
|
563 | 594 | ;;;;;;;;;; |
564 | | - :else |
565 | | - (recur (+ offset result))))) |
| 595 | + :else |
| 596 | + (recur (+ offset result))))) |
566 | 597 |
|
567 | | - (.clear ^ByteBuffer buf) |
568 | | - (recur 0))) |
| 598 | + (.clear ^ByteBuffer buf) |
| 599 | + (recur 0)) |
569 | 600 |
|
570 | 601 | ;;;;;;;;;; |
571 | 602 | :else |
|
0 commit comments