Merge remote-tracking branch 'origin/master' into 25-input-validation…

…-is-broken-or-incomplete-in-quite-a-few-forms

docker-compose-dev.yml

@@ -59,7 +59,6 @@ services:
 
  traefik:
  container_name: pm_traefik
- image: traefik
  build:
  context: ./dockers/traefik
  args:

docker-compose.yml

@@ -59,7 +59,6 @@ services:
 
  traefik:
  container_name: pm_traefik
- image: traefik
  build:
  context: ./dockers/traefik
  args:
@@ -108,7 +107,7 @@ services:
 ##
  manager:
  container_name: pm_manager
- image: "pm_manager"
+ image: "pm-manager"
  build:
  context: ./dockers/manager
  depends_on:

dockers/manager/back/app/docker/__init__.py

@@ -22,5 +22,5 @@ def formatTime(time):
 from . import volume
 from . import shell
 from . import logs
-from . import image_pull
+from . import streams
 from . import files

dockers/manager/back/app/docker/image.py

@@ -9,6 +9,7 @@
 import asyncio
 import re
 from app.exception import PMException
+from .streams import StreamFollower
 
 DockerImage = createType("DockerImage")
 
@@ -201,11 +202,39 @@ def resolve_prune_images(*_, onlyDangling):
  "spaceReclaimed": pruned.get("SpaceReclaimed", None),
  }
 
+
 @registerMutation("tagDockerImage")
 def resolve_tag_image(*_, id, repository, tag=None):
  try:
  docker_client.images.get(id).tag(repository, tag)
  except APIError as e:
  raise PMException(e.explanation)
  except Exception as e:
- raise PMException(str(e))
+ raise PMException(str(e))
+
+
+@registerMutation("buildDockerImage")
+def build_docker_image(*_, input):
+ url = input["url"]
+ tag = input["tag"]
+
+ try:
+ stream = docker_client.api.build(url, tag, decode=True, rm=True, forcerm=True)
+ return StreamFollower.create(stream, {"name": tag, "type": "BUILD"})
+ except APIError as e:
+ raise PMException(e.explanation)
+ except Exception as e:
+ raise PMException(str(e))
+
+
+@registerMutation("pullDockerImage")
+async def mutation_pull_image(*_, name):
+ name = name if ":" in name else f"{name}:latest"
+ try:
+ stream = docker_client.api.pull(name, stream=True, decode=True)
+ return StreamFollower.create(stream, {"name": name, "type": "PULL"})
+ except APIError as e:
+ raise PMException(e.explanation)
+ except Exception as e:
+ raise PMException(str(e))
+

dockers/manager/back/app/docker/streams.py

@@ -0,0 +1,117 @@
+import docker
+import asyncio
+import random
+from uuid import uuid4
+from dataclasses import dataclass, field
+from collections import defaultdict
+
+from app.api import docker_client
+from app.utils import registerMutation, registerSubscription, registerQuery
+from . import kv_to_dict
+
+
+@dataclass
+class Runner:
+ meta: dict
+ queues: set[asyncio.Queue] = field(default_factory=set)
+ messages: list = field(default_factory=list)
+ done: bool = False
+
+
+class StreamFollower:
+ runners: dict[str, Runner] = {}
+ subscribers: set[asyncio.Queue] = set()
+
+ @classmethod
+ async def follow(cls, id):
+ queue = asyncio.Queue()
+ runner = cls.runners.get(id)
+ runner.queues.add(queue)
+
+ for message in runner.messages:
+ yield message
+ if runner.done:
+ return
+ while message := await queue.get():
+ yield message
+ queue.task_done()
+
+ @classmethod
+ def has(cls, id, check=None):
+ if id in cls.runners:
+ if check is None:
+ return True
+ return check(cls.runners[id].meta)
+ return False
+
+ @classmethod
+ async def subscribe(cls):
+ queue = asyncio.Queue()
+ cls.subscribers.add(queue)
+ while True:
+ start_info = await queue.get()
+ yield start_info
+ queue.task_done()
+
+ @classmethod
+ def filter(cls, filter=lambda x: True):
+ return [runner for runner in cls.runners.values() if filter(runner.meta)]
+
+ @classmethod
+ def create(cls, stream, meta):
+ id = str(uuid4())
+ meta = {**meta, "id": id}
+ runner = Runner(meta)
+ cls.runners[id] = runner
+
+ async def start():
+ try:
+ for message in stream:
+ runner.messages.append(message)
+ for queue in runner.queues:
+ await queue.put(message)
+ await asyncio.sleep(0) # release loop
+ except docker.errors.APIError as e:
+ runner.messages.append({"error": e.explanation})
+ except Exception as e:
+ runner.messages.append({"error": str(e)})
+ runner.done = True
+
+ await asyncio.sleep(10) # Keep the task 10s
+ cls.runners.pop(id)
+
+ for sub in cls.subscribers:
+ sub.put_nowait(meta)
+ asyncio.create_task(start())
+ return meta
+
+
+@registerQuery("dockerStreams")
+def resolve_streams(*_):
+ return [r.meta for r in StreamFollower.runners.values()]
+
+
+@registerSubscription("dockerStreamPull")
+async def subscribe_to_pull(*_, id):
+ if not StreamFollower.has(id, lambda r: r["type"] == "PULL"):
+ yield {"done": True}
+ return
+ async for message in StreamFollower.follow(id):
+ yield {**message, "done": False}
+ yield {"done": True}
+
+
+@registerSubscription("dockerStreamBuild")
+async def subscribe_to_build(*_, id):
+ if not StreamFollower.has(id, lambda r: r["type"] == "BUILD"):
+ yield {"done": True}
+ return
+ async for message in StreamFollower.follow(id):
+ yield {**message, "done": False}
+ yield {"done": True}
+
+
+@registerSubscription("dockerStreamStart")
+async def resolve_progress_start(*_):
+ async for event in StreamFollower.subscribe():
+ yield event

dockers/manager/back/schema/docker/image.graphql

@@ -9,13 +9,18 @@ extend type Mutation {
  force: Boolean! = false
  pruneParents: Boolean! = true
  ): BasicMutationResponse!
- tagDockerImage(id: ID!, repository: String!, tag: String): BasicMutationResponse!
+ tagDockerImage(
+ id: ID!
+ repository: String!
+ tag: String
+ ): BasicMutationResponse!
  pruneDockerImages(onlyDangling: Boolean! = true): PruneResponse!
+ buildDockerImage(input: BuildDockerImageInput): DockerImageBuildResponse!
 }
 
 type DockerImageHistoryEntry {
- operation: String !
- argument: String !
+ operation: String!
+ argument: String!
  comment: String!
  tags: [String!]!
  date: DateTime
@@ -76,21 +81,20 @@ type DockerSearchImageTagResult {
  lastUpdated: DateTime!
 }
 
-type DockerImagePullProgressDetail {
- current: Int
- total: Int
-}
 
-type DockerImagePullLogEntry {
- status: String!
- id: String
- progressDetail: DockerImagePullProgressDetail
- progress: String
+input BuildDockerImageInput {
+ url: String!
+ tag: String!
 }
 
 type DockerImagePull {
  id: ID!
- name: String
+ name: String!
+}
+
+type DockerImageBuild {
+ id: ID!
+ name: String!
 }
 
 type DockerImagePullResponse implements IMutationResponse {
@@ -99,10 +103,8 @@ type DockerImagePullResponse implements IMutationResponse {
  result: DockerImagePull
 }
 
-extend type Query {
- dockerImagePulls: [DockerImagePull!]!
-}
-
-extend type Subscription {
- pullImageProgress(id: ID!): DockerImagePullLogEntry
+type DockerImageBuildResponse implements IMutationResponse {
+ success: Boolean!
+ error: String
+ result: DockerImageBuild
 }

dockers/manager/back/schema/docker/stream.graphql

@@ -0,0 +1,35 @@
+type DockerImagePullProgressDetail {
+ current: Int
+ total: Int
+}
+
+type DockerStreamPullLogEntry {
+ done: Boolean!
+ error: String
+ status: String
+ id: String
+ progressDetail: DockerImagePullProgressDetail
+ progress: String
+}
+
+type DockerSteamBuildLogEntry {
+ error: String
+ done: Boolean!
+ stream: String
+}
+
+type DockerStream {
+ id: ID!
+ type: String!
+ name: String!
+}
+
+extend type Query {
+ dockerStreams: [DockerStream!]!
+}
+
+extend type Subscription {
+ dockerStreamStart: DockerStream!
+ dockerStreamPull(id: ID!): DockerStreamPullLogEntry
+ dockerStreamBuild(id: ID!): DockerSteamBuildLogEntry
+}

dockers/manager/front/src/api/docker/images/index.js

@@ -1,11 +1,9 @@
 import * as fragments from "./fragments.js";
 import * as mutations from "./mutations.js";
 import * as queries from "./queries.js";
-import * as subscriptions from "./subscriptions.js";
 
 export default {
  ...fragments,
  ...mutations,
- ...subscriptions,
  ...queries
 };