Merge pull request #15 from sutr90/feature/support-go-docker-client

Feature/support go docker client

Author: ivictbor

.vscode/launch.json

@@ -0,0 +1,13 @@
+{
+    "version": "0.2.0",
+    "configurations": [
+        
+        {
+            "name": "Launch Package",
+            "type": "go",
+            "request": "launch",
+            "mode": "auto",
+            "program": "${fileDirname}"
+        }
+    ]
+}

README.md

@@ -80,6 +80,7 @@ Once done, just go to <your host> and login as "admin" with <any password>.
 ## Available Environment Options:
 | Environment Variable       | Description   | Defaults | Required |
 |----------------------------|---------------------------------|--------|-----------------|
+| DOCKER_HOST              | URL of the docker socket to connect to. See below | `unix:///var/run/docker.sock` | |
 | ADMIN_USERNAME           | Username for initial user                        | `admin`                 | if `AGENT=false`
 | ADMIN_PASSWORD           | Password for initial user                        |                    | if `AGENT=false`
 | PORT               | Port to listen on                                | `2874`             | if `AGENT=false`
@@ -89,3 +90,8 @@ Once done, just go to <your host> and login as "admin" with <any password>.
 | HOST               | Url to OnLogs host from protocol to domain name. |                    | if `AGENT=true`
 | ONLOGS_TOKEN       | Token that will use an agent to authorize and connect to HOST | Generates with OnLogs interface   | if `AGENT=true`
 | MAX_LOGS_SIZE | Maximum allowed total logs size before cleanup triggers. Accepts human-readable formats like 5GB, 500MB, 1.5GB etc. When exceeded, 10% of logs (by count) will be removed proportionally across containers starting from oldest | 10GB | -
+
+### Docket socket URL
+By default the app will connect using the raw unix socket. But this can be overriden via the ENV variable `DOCKER_HOST`. That way you can specify fully qualified URL to the socket or URL of an docker socket proxy.
+
+In `compose-socket-proxy.yml` you can see a sample compose file for starting the socket proxy. To use it in the app set `DOCKER_HOST=http://localhost:2375` in the ENV.

application/backend/app/daemon/daemon.go

@@ -2,22 +2,28 @@ package daemon
 
 import (
 	"bufio"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
-	"net"
 	"os"
 	"strconv"
 	"strings"
 	"time"
 
 	"github.com/devforth/OnLogs/app/agent"
 	"github.com/devforth/OnLogs/app/containerdb"
+	"github.com/devforth/OnLogs/app/docker"
 	"github.com/devforth/OnLogs/app/util"
 	"github.com/devforth/OnLogs/app/vars"
+	"github.com/docker/docker/api/types/container"
 	"github.com/syndtr/goleveldb/leveldb"
 )
 
+type DaemonService struct {
+	DockerClient *docker.DockerService
+}
+
 func createLogMessage(db *leveldb.DB, host string, container string, message string) string {
 	datetime := strings.Replace(strings.Split(time.Now().UTC().String(), " +")[0], " ", "T", 1)
 	if len(datetime) < 29 {
@@ -40,25 +46,6 @@ func validateMessage(message string) (string, bool) {
 	return message, true
 }
 
-func createConnection(containerName string) net.Conn {
-	connection, _ := net.Dial("unix", os.Getenv("DOCKER_SOCKET_PATH"))
-	fmt.Fprintf(
-		connection,
-		"GET /containers/"+containerName+"/logs?stdout=true&stderr=true&timestamps=true&follow=true&since="+strconv.FormatInt(time.Now().Add(-5*time.Second).Unix(), 10)+" HTTP/1.0\r\n\r\n",
-	)
-	return connection
-}
-
-func readHeader(reader bufio.Reader) {
-	for { // reading resp header
-		tmp, _ := reader.ReadString('\n')
-		if tmp[:len(tmp)-2] == "" {
-			reader.ReadString('\n')
-			break
-		}
-	}
-}
-
 func closeActiveStream(containerName string) {
 	newDaemonStreams := []string{}
 	for _, stream := range vars.Active_Daemon_Streams {
@@ -73,19 +60,29 @@ func closeActiveStream(containerName string) {
 	vars.Active_Daemon_Streams = newDaemonStreams
 }
 
-func CreateDaemonToHostStream(containerName string) {
-	connection := createConnection(containerName)
-	reader := bufio.NewReader(connection)
-	readHeader(*reader)
+func (h *DaemonService) CreateDaemonToHostStream(ctx context.Context, containerName string) {
+	rc, err := h.DockerClient.Client.ContainerLogs(ctx, containerName, container.LogsOptions{ShowStdout: true, ShowStderr: true, Timestamps: true, Follow: true, Since: strconv.FormatInt(time.Now().Add(-5*time.Second).Unix(), 10)})
+	if err != nil {
+		closeActiveStream(containerName)
+		return
+	}
+	defer rc.Close()
+
+	reader := bufio.NewReader(rc)
 
 	host := util.GetHost()
 	token := os.Getenv("ONLOGS_TOKEN")
 	agent.SendLogMessage(token, containerName, strings.SplitN(createLogMessage(nil, host, containerName, "ONLOGS: Container listening started!"), " ", 2))
 
 	lastSleep := time.Now().Unix()
 	for { // reading body
-		logLine, get_string_error := reader.ReadString('\n') // TODO read bytes instead of strings
+		logLine, get_string_error := reader.ReadString('\n')
 		if get_string_error != nil {
+			if get_string_error == io.EOF {
+				closeActiveStream(containerName)
+				agent.SendLogMessage(token, containerName, strings.SplitN(createLogMessage(nil, host, containerName, "ONLOGS: Container listening stopped! (EOF)"), " ", 2))
+				return
+			}
 			closeActiveStream(containerName)
 			agent.SendLogMessage(token, containerName, strings.SplitN(createLogMessage(nil, host, containerName, "ONLOGS: Container listening stopped! ("+get_string_error.Error()+")"), " ", 2))
 			return
@@ -106,10 +103,15 @@ func CreateDaemonToHostStream(containerName string) {
 }
 
 // creates stream that writes logs from every docker container to leveldb
-func CreateDaemonToDBStream(containerName string) {
-	connection := createConnection(containerName)
-	reader := bufio.NewReader(connection)
-	readHeader(*reader)
+func (h *DaemonService) CreateDaemonToDBStream(ctx context.Context, containerName string) {
+	rc, err := h.DockerClient.Client.ContainerLogs(ctx, containerName, container.LogsOptions{ShowStdout: true, ShowStderr: true, Timestamps: true, Follow: true, Since: strconv.FormatInt(time.Now().Add(-5*time.Second).Unix(), 10)})
+	if err != nil {
+		closeActiveStream(containerName)
+		return
+	}
+	defer rc.Close()
+
+	reader := bufio.NewReader(rc)
 
 	host := util.GetHost()
 	current_db := util.GetDB(host, containerName, "logs")
@@ -119,6 +121,11 @@ func CreateDaemonToDBStream(containerName string) {
 	for { // reading body
 		logLine, get_string_error := reader.ReadString('\n')
 		if get_string_error != nil {
+			if get_string_error == io.EOF {
+				closeActiveStream(containerName)
+				createLogMessage(current_db, host, containerName, "ONLOGS: Container listening stopped! (EOF)")
+				return
+			}
 			closeActiveStream(containerName)
 			createLogMessage(current_db, host, containerName, "ONLOGS: Container listening stopped! ("+get_string_error.Error()+")")
 			return
@@ -150,27 +157,12 @@ func CreateDaemonToDBStream(containerName string) {
 	}
 }
 
-// make request to docker socket
-func makeSocketRequest(path string) []byte {
-	connection, err := net.Dial("unix", os.Getenv("DOCKER_SOCKET_PATH"))
+// returns list of names of docker containers from docker daemon
+func (h *DaemonService) GetContainersList(ctx context.Context) []string {
+	result, err := h.DockerClient.GetContainerNames(ctx)
 	if err != nil {
 		panic(err)
 	}
-	fmt.Fprintf(connection, "GET /"+path+" HTTP/1.0\r\n\r\n")
-
-	body, _ := io.ReadAll(connection)
-
-	connection.Close()
-	return body
-}
-
-// returns list of names of docker containers from docker daemon
-func GetContainersList() []string {
-	var result []map[string]any
-
-	body := string(makeSocketRequest("containers/json"))
-	body = strings.Split(body, "\r\n\r\n")[1]
-	json.Unmarshal([]byte(body), &result)
 
 	var names []string
 
@@ -184,9 +176,9 @@ func GetContainersList() []string {
 	}
 	containersMetaDB = vars.ContainersMeta_DBs[util.GetHost()]
 
-	for i := 0; i < len(result); i++ {
-		name := fmt.Sprintf("%v", result[i]["Names"].([]interface{})[0].(string))[1:]
-		id := result[i]["Id"].(string)
+	for i := range result {
+		name := result[i].Name
+		id := result[i].ID
 
 		names = append(names, name)
 		containersMetaDB.Put([]byte(name), []byte(id), nil)
@@ -195,16 +187,11 @@ func GetContainersList() []string {
 	return names
 }
 
-func GetContainerImageNameByContainerID(containerID string) string {
-	body := string(makeSocketRequest("containers/" + containerID + "/json"))
-	body = strings.Split(body, "\r\n\r\n")[1]
-	var result map[string]any
-	json.Unmarshal([]byte(body), &result)
-
-	if result["Config"] == nil {
+func (h *DaemonService) GetContainerImageNameByContainerID(ctx context.Context, containerID string) string {
+	result, err := h.DockerClient.GetContainerImageNameByContainerID(ctx, containerID)
+	if err != nil {
 		return ""
 	}
 
-	imageName := fmt.Sprintf("%v", result["Config"].(map[string]any)["Image"])
-	return imageName
+	return result
 }

application/backend/app/docker/docker_client.go

@@ -0,0 +1,45 @@
+package docker
+
+import (
+	"context"
+	"strings"
+
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/client"
+)
+
+type DockerService struct {
+    Client *client.Client
+}
+
+type ContainerNamesResult struct {
+	Name string
+	ID string
+}
+
+func (s *DockerService) GetContainerNames(ctx context.Context) ([]ContainerNamesResult, error) {
+    containers, err := s.Client.ContainerList(ctx, container.ListOptions{All: true})
+    if err != nil {
+        return nil, err
+    }
+    
+    var res []ContainerNamesResult
+    for _, c := range containers {
+        name := ""
+        if len(c.Names) > 0 {
+            name = strings.TrimPrefix(c.Names[0], "/")
+        }
+        res = append(res, ContainerNamesResult{Name: name, ID: c.ID})
+    }
+    return res, nil
+}
+
+
+func (s *DockerService) GetContainerImageNameByContainerID(ctx context.Context, containerID string) (string, error) {
+    res, err := s.Client.ContainerInspect(ctx, containerID)
+    if err != nil {
+        return "", err
+    }
+    
+    return res.Config.Image, nil
+}