Rename to moby-ryuk, replace HTTP based scheduling with TCP connection

Dockerfile

@@ -1,11 +1,11 @@
 FROM golang:1.9 as workspace
-WORKDIR /go/src/github.com/bsideup/moby-massacrer
+WORKDIR /go/src/github.com/bsideup/moby-ryuk
 COPY glide.lock glide.yaml Makefile ./
 RUN make bootstrap
 COPY . .
 RUN make build
 
 FROM alpine:latest
 RUN apk --no-cache add ca-certificates
-COPY --from=workspace /go/src/github.com/bsideup/moby-massacrer/bin/moby-massacrer /app
+COPY --from=workspace /go/src/github.com/bsideup/moby-ryuk/bin/moby-ryuk /app
 CMD ["/app"]

Makefile

@@ -2,7 +2,7 @@
 
 SOURCE_FOLDER := .
 
-BINARY_PATH ?= ./bin/moby-massacrer
+BINARY_PATH ?= ./bin/moby-ryuk
 
 GOARCH ?= amd64
 
@@ -20,6 +20,9 @@ build_all: vet fmt
 compile:
 	CGO_ENABLED=0 go build -i -v -ldflags '-s' -o $(BINARY_PATH) $(SOURCE_FOLDER)/
 
+run:
+	go run $(SOURCE_FOLDER)/main.go
+
 build: vet fmt compile
 
 fmt:

README.md

@@ -1,23 +1,36 @@
-# Moby Massacrer
+# Moby Ryuk
 
 This project helps you to remove containers/networks/volumes by given filter after specified delay.
 
 # Usage
 
 1. Start it:
 
-        $ ./bin/moby-massacrer -p 8080
+        $ ./bin/moby-ryuk -p 8080
         $ # You can also run it with Docker
-        $ docker run -v /var/run/docker.sock:/var/run/docker.sock -p 8080:8080 bsideup/moby-massacrer
+        $ docker run -v /var/run/docker.sock:/var/run/docker.sock -p 8080:8080 bsideup/moby-ryuk
 
-1. Submit cleanup request:
+1. Connect via TCP:
 
-        $ curl -d "label=testing=true" -d "health=unhealthy" http://localhost:8080/schedule?delay=1h
+        $ nc localhost 8080
 
-1. Realize that 1 hour is too long for the demo and change it to 5 seconds:
+1. Send some filters:
 
-        $ curl -d "label=testing=true" -d "health=unhealthy" http://localhost:8080/schedule?delay=5s
+        label=testing=true&health=unhealthy
+        ACK
+        label=something
+        ACK
 
-1. See containers/networks/volumes deleted after 5s:
+1. Close the connection
 
-        Removed 1 container(s), 0 network(s), 0 volume(s)
+1. Send more filters with "one-off" style:
+
+        printf "label=something_else" | nc localhost 8080
+
+1. See containers/networks/volumes deleted after 10s:
+
+        2018/01/15 18:38:52 Timed out waiting for connection
+        2018/01/15 18:38:52 Deleting {"label":{"something":true}}
+        2018/01/15 18:38:52 Deleting {"label":{"something_else":true}}
+        2018/01/15 18:38:52 Deleting {"health":{"unhealthy":true},"label":{"testing=true":true}}
+        2018/01/15 18:38:52 Removed 1 container(s), 0 network(s), 0 volume(s)

glide.yaml

@@ -1,4 +1,4 @@
-package: github.com/bsideup/moby-massacrer
+package: github.com/bsideup/moby-ryuk
 import:
 - package: github.com/docker/docker
   version: v17.05.0-ce-rc3

main.go

@@ -1,12 +1,13 @@
 package main
 
 import (
+	"bufio"
 	"context"
 	"flag"
 	"fmt"
 	"log"
-	"mime"
-	"net/http"
+	"net"
+	"net/url"
 	"time"
 
 	"github.com/docker/docker/api/types"
@@ -20,124 +21,111 @@ func main() {
 	flag.Parse()
 	log.Printf("Starting on port %d...", *port)
 
-	schedule := make(map[string]time.Time)
-
 	cli, err := client.NewEnvClient()
 	if err != nil {
 		panic(err)
 	}
 
-	go runMassacre(&schedule, cli)
+	deathNote := make(map[string]bool)
 
-	mux := http.NewServeMux()
-	mux.HandleFunc("/schedule", func(w http.ResponseWriter, r *http.Request) {
-		if r.Method != "POST" {
-			http.Error(w, "Only 'POST' method is allowed", http.StatusMethodNotAllowed)
-			return
-		}
+	connected := make(chan bool)
+	disconnected := make(chan bool)
 
-		mediaType, _, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
+	go func() {
+		ln, _ := net.Listen("tcp", fmt.Sprintf(":%d", *port))
+		for {
+			conn, _ := ln.Accept()
+			connected <- true
+			reader := bufio.NewReader(conn)
+			for {
+				message, err := reader.ReadString('\n')
 
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusUnsupportedMediaType)
-			return
-		}
+				if len(message) > 0 {
+					query, err := url.ParseQuery(message)
 
-		if mediaType != "application/x-www-form-urlencoded" {
-			http.Error(w, "Only 'application/x-www-form-urlencoded' content type is allowed", http.StatusUnsupportedMediaType)
-			return
-		}
+					if err != nil {
+						log.Println(err)
+						continue
+					}
 
-		err = r.ParseForm()
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusBadRequest)
-			return
-		}
+					args := filters.NewArgs()
+					for filterType, values := range query {
+						for _, value := range values {
+							args.Add(filterType, value)
+						}
+					}
+					param, err := filters.ToParam(args)
 
-		args := filters.NewArgs()
+					if err != nil {
+						log.Println(err)
+						continue
+					}
 
-		for filterType, values := range r.PostForm {
-			for _, value := range values {
-				args.Add(filterType, value)
-			}
-		}
+					log.Printf("%+v\n", param)
 
-		if args.Len() <= 0 {
-			http.Error(w, "Empty filters", http.StatusBadRequest)
-			return
-		}
+					deathNote[param] = true
 
-		delay, err := time.ParseDuration(r.URL.Query().Get("delay"))
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusBadRequest)
-			return
-		}
+					conn.Write([]byte("ACK\n"))
+				}
 
-		param, err := filters.ToParam(args)
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusInternalServerError)
-			return
+				if err != nil {
+					log.Println(err)
+					break
+				}
+			}
+			disconnected <- true
+			conn.Close()
 		}
-		at := time.Now().Add(delay)
-		log.Printf("Scheduling %s at %s", param, at)
-		schedule[param] = at
+	}()
 
-		w.WriteHeader(http.StatusAccepted)
-	})
-
-	log.Fatal(http.ListenAndServe(fmt.Sprintf(":%d", *port), mux))
-}
-
-func runMassacre(schedule *map[string]time.Time, cli *client.Client) {
-	var now time.Time
+TimeoutLoop:
 	for {
-		now = time.Now()
-		oldSchedule := *schedule
-
-		*schedule = make(map[string]time.Time)
-
-		deletedContainers := make(map[string]bool)
-		deletedNetworks := make(map[string]bool)
-		deletedVolumes := make(map[string]bool)
-
-		for param, after := range oldSchedule {
-			if now.Before(after) {
-				(*schedule)[param] = after
-				continue
+		select {
+		case <-connected:
+			log.Println("Connected")
+		case <-disconnected:
+			log.Println("Disconnected")
+			select {
+			case <-connected:
+			case <-time.After(10 * time.Second):
+				log.Println("Timed out waiting for connection")
+				break TimeoutLoop
 			}
-			log.Printf("Deleting %s after %s\n", param, after)
+		}
+	}
 
-			args, err := filters.FromParam(param)
-			if err != nil {
-				log.Println(err)
-				continue
-			}
+	deletedContainers := make(map[string]bool)
+	deletedNetworks := make(map[string]bool)
+	deletedVolumes := make(map[string]bool)
 
-			containers, err := cli.ContainerList(context.Background(), types.ContainerListOptions{All: true, Filters: args})
-			if err != nil {
-				log.Println(err)
-			} else {
-				for _, container := range containers {
-					cli.ContainerRemove(context.Background(), container.ID, types.ContainerRemoveOptions{RemoveVolumes: true, Force: true})
-					deletedContainers[container.ID] = true
-				}
-			}
+	for param := range deathNote {
+		log.Printf("Deleting %s\n", param)
 
-			networksPruneReport, err := cli.NetworksPrune(context.Background(), args)
-			for _, networkID := range networksPruneReport.NetworksDeleted {
-				deletedNetworks[networkID] = true
-			}
+		args, err := filters.FromParam(param)
+		if err != nil {
+			log.Println(err)
+			continue
+		}
 
-			volumesPruneReport, err := cli.VolumesPrune(context.Background(), args)
-			for _, volumeName := range volumesPruneReport.VolumesDeleted {
-				deletedVolumes[volumeName] = true
+		if containers, err := cli.ContainerList(context.Background(), types.ContainerListOptions{All: true, Filters: args}); err != nil {
+			log.Println(err)
+		} else {
+			for _, container := range containers {
+				cli.ContainerRemove(context.Background(), container.ID, types.ContainerRemoveOptions{RemoveVolumes: true, Force: true})
+				deletedContainers[container.ID] = true
 			}
 		}
-		if len(deletedContainers)+len(deletedNetworks)+len(deletedVolumes) <= 0 {
-			time.Sleep(time.Second)
-		} else {
-			log.Printf("Removed %d container(s), %d network(s), %d volume(s)", len(deletedContainers), len(deletedNetworks), len(deletedVolumes))
+
+		networksPruneReport, err := cli.NetworksPrune(context.Background(), args)
+		for _, networkID := range networksPruneReport.NetworksDeleted {
+			deletedNetworks[networkID] = true
 		}
 
+		volumesPruneReport, err := cli.VolumesPrune(context.Background(), args)
+		for _, volumeName := range volumesPruneReport.VolumesDeleted {
+			deletedVolumes[volumeName] = true
+		}
 	}
+
+	log.Printf("Removed %d container(s), %d network(s), %d volume(s)", len(deletedContainers), len(deletedNetworks), len(deletedVolumes))
 }