implement godoxy-agent

agent/pkg/handler/check_health.go

@@ -0,0 +1,66 @@
+package handler
+
+import (
+	"net/http"
+	"net/url"
+
+	apiUtils "github.com/yusing/go-proxy/internal/api/v1/utils"
+	"github.com/yusing/go-proxy/internal/net/types"
+	"github.com/yusing/go-proxy/internal/utils"
+	"github.com/yusing/go-proxy/internal/watcher/health"
+	"github.com/yusing/go-proxy/internal/watcher/health/monitor"
+)
+
+func CheckHealth(w http.ResponseWriter, r *http.Request) {
+	query := r.URL.Query()
+	scheme := query.Get("scheme")
+	if scheme == "" {
+		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+		return
+	}
+
+	var result *health.HealthCheckResult
+	var err error
+	switch scheme {
+	case "fileserver":
+		path := query.Get("path")
+		if path == "" {
+			http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+			return
+		}
+		ok, err := utils.FileExists(path)
+		result = &health.HealthCheckResult{Healthy: ok}
+		if err != nil {
+			result.Detail = err.Error()
+		}
+	case "http", "https": // path is optional
+		host := query.Get("host")
+		path := query.Get("path")
+		if host == "" {
+			http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+			return
+		}
+		result, err = monitor.NewHTTPHealthChecker(types.NewURL(&url.URL{
+			Scheme: scheme,
+			Host:   host,
+			Path:   path,
+		}), health.DefaultHealthConfig).CheckHealth()
+	case "tcp", "udp":
+		host := query.Get("host")
+		if host == "" {
+			http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+			return
+		}
+		result, err = monitor.NewRawHealthChecker(types.NewURL(&url.URL{
+			Scheme: scheme,
+			Host:   host,
+		}), health.DefaultHealthConfig).CheckHealth()
+	}
+
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadGateway)
+		return
+	}
+
+	apiUtils.RespondJSON(w, r, result)
+}

agent/pkg/handler/docker_socket.go

@@ -0,0 +1,92 @@
+package handler
+
+import (
+	"bufio"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+
+	"github.com/yusing/go-proxy/internal/common"
+	"github.com/yusing/go-proxy/internal/docker"
+	"github.com/yusing/go-proxy/internal/logging"
+	godoxyIO "github.com/yusing/go-proxy/internal/utils"
+)
+
+func DockerSocketHandler() http.HandlerFunc {
+	dockerClient, err := docker.ConnectClient(common.DockerHostFromEnv)
+	if err != nil {
+		logging.Fatal().Err(err).Msg("failed to connect to docker client")
+	}
+	dockerDialerCallback := dockerClient.Dialer()
+
+	return func(w http.ResponseWriter, r *http.Request) {
+		conn, err := dockerDialerCallback(r.Context())
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		defer conn.Close()
+
+		// Create a done channel to handle cancellation
+		done := make(chan struct{})
+		defer close(done)
+
+		closed := false
+
+		// Start a goroutine to monitor context cancellation
+		go func() {
+			select {
+			case <-r.Context().Done():
+				closed = true
+				conn.Close() // Force close the connection when client disconnects
+			case <-done:
+			}
+		}()
+
+		if err := r.Write(conn); err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+
+		resp, err := http.ReadResponse(bufio.NewReader(conn), r)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+		defer resp.Body.Close()
+
+		// Set any response headers before writing the status code
+		for k, v := range resp.Header {
+			w.Header()[k] = v
+		}
+		w.WriteHeader(resp.StatusCode)
+
+		// For event streams, we need to flush the writer to ensure
+		// events are sent immediately
+		if f, ok := w.(http.Flusher); ok && strings.HasSuffix(r.URL.Path, "/events") {
+			// Copy the body in chunks and flush after each write
+			buf := make([]byte, 2048)
+			for {
+				n, err := resp.Body.Read(buf)
+				if n > 0 {
+					_, werr := w.Write(buf[:n])
+					if werr != nil {
+						logging.Error().Err(werr).Msg("error writing docker event response")
+						break
+					}
+					f.Flush()
+				}
+				if err != nil {
+					if !closed && !errors.Is(err, io.EOF) {
+						logging.Error().Err(err).Msg("error reading docker event response")
+					}
+					return
+				}
+			}
+		} else {
+			// For non-event streams, just copy the body
+			godoxyIO.NewPipe(r.Context(), resp.Body, NopWriteCloser{w}).Start()
+		}
+	}
+}

agent/pkg/handler/handler.go

@@ -0,0 +1,50 @@
+package handler
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+
+	"github.com/yusing/go-proxy/agent/pkg/agent"
+	"github.com/yusing/go-proxy/agent/pkg/env"
+	v1 "github.com/yusing/go-proxy/internal/api/v1"
+	"github.com/yusing/go-proxy/internal/logging/memlogger"
+	"github.com/yusing/go-proxy/internal/utils/strutils"
+)
+
+type ServeMux struct{ *http.ServeMux }
+
+func (mux ServeMux) HandleMethods(methods, endpoint string, handler http.HandlerFunc) {
+	for _, m := range strutils.CommaSeperatedList(methods) {
+		mux.ServeMux.HandleFunc(m+" "+agent.APIEndpointBase+endpoint, handler)
+	}
+}
+
+func (mux ServeMux) HandleFunc(endpoint string, handler http.HandlerFunc) {
+	mux.ServeMux.HandleFunc(agent.APIEndpointBase+endpoint, handler)
+}
+
+type NopWriteCloser struct {
+	io.Writer
+}
+
+func (NopWriteCloser) Close() error {
+	return nil
+}
+
+func NewHandler(caCertPEM []byte) http.Handler {
+	mux := ServeMux{http.NewServeMux()}
+
+	mux.HandleFunc(agent.EndpointProxyHTTP+"/{path...}", ProxyHTTP)
+	mux.HandleMethods("GET", agent.EndpointVersion, v1.GetVersion)
+	mux.HandleMethods("GET", agent.EndpointName, func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprint(w, env.AgentName)
+	})
+	mux.HandleMethods("GET", agent.EndpointCACert, func(w http.ResponseWriter, r *http.Request) {
+		w.Write(caCertPEM)
+	})
+	mux.HandleMethods("GET", agent.EndpointHealth, CheckHealth)
+	mux.HandleMethods("GET", agent.EndpointLogs, memlogger.LogsWS(nil))
+	mux.ServeMux.HandleFunc("/", DockerSocketHandler())
+	return mux
+}

agent/pkg/handler/proxy_http.go

@@ -0,0 +1,59 @@
+package handler
+
+import (
+	"crypto/tls"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/yusing/go-proxy/agent/pkg/agent"
+	agentproxy "github.com/yusing/go-proxy/agent/pkg/agentproxy"
+	"github.com/yusing/go-proxy/internal/logging"
+	gphttp "github.com/yusing/go-proxy/internal/net/http"
+	"github.com/yusing/go-proxy/internal/net/http/reverseproxy"
+	"github.com/yusing/go-proxy/internal/net/types"
+	"github.com/yusing/go-proxy/internal/utils/strutils"
+)
+
+func ProxyHTTP(w http.ResponseWriter, r *http.Request) {
+	host := r.Header.Get(agentproxy.HeaderXProxyHost)
+	isHTTPs := strutils.ParseBool(r.Header.Get(agentproxy.HeaderXProxyHTTPS))
+	skipTLSVerify := strutils.ParseBool(r.Header.Get(agentproxy.HeaderXProxySkipTLSVerify))
+	responseHeaderTimeout, err := strconv.Atoi(r.Header.Get(agentproxy.HeaderXProxyResponseHeaderTimeout))
+	if err != nil {
+		responseHeaderTimeout = 0
+	}
+
+	logging.Debug().Msgf("proxy http request: host=%s, isHTTPs=%t, skipTLSVerify=%t, responseHeaderTimeout=%d", host, isHTTPs, skipTLSVerify, responseHeaderTimeout)
+
+	if host == "" {
+		http.Error(w, "missing required headers", http.StatusBadRequest)
+		return
+	}
+
+	scheme := "http"
+	if isHTTPs {
+		scheme = "https"
+	}
+
+	var transport *http.Transport
+	if skipTLSVerify {
+		transport = gphttp.NewTransportWithTLSConfig(&tls.Config{InsecureSkipVerify: true})
+	} else {
+		transport = gphttp.NewTransport()
+	}
+
+	if responseHeaderTimeout > 0 {
+		transport = transport.Clone()
+		transport.ResponseHeaderTimeout = time.Duration(responseHeaderTimeout) * time.Second
+	}
+
+	r.URL.Scheme = scheme
+	r.URL.Host = host
+	r.URL.Path = r.URL.Path[agent.HTTPProxyURLStripLen:] // strip the {API_BASE}/proxy/http prefix
+
+	logging.Debug().Msgf("proxy http request: %s %s", r.Method, r.URL.String())
+
+	rp := reverseproxy.NewReverseProxy("agent", types.NewURL(r.URL), transport)
+	rp.ServeHTTP(w, r)
+}