feat: godoxy agent

agent/pkg/handler/check_health.go

@@ -0,0 +1,78 @@
+package handler
+
+import (
+	"fmt"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+
+	"github.com/yusing/go-proxy/internal/net/gphttp"
+	"github.com/yusing/go-proxy/internal/net/types"
+	"github.com/yusing/go-proxy/internal/watcher/health"
+	"github.com/yusing/go-proxy/internal/watcher/health/monitor"
+)
+
+var defaultHealthConfig = health.DefaultHealthConfig()
+
+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
+		}
+		_, err := os.Stat(path)
+		result = &health.HealthCheckResult{Healthy: err == nil}
+		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,
+		}), defaultHealthConfig).CheckHealth()
+	case "tcp", "udp":
+		host := query.Get("host")
+		if host == "" {
+			http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+			return
+		}
+		hasPort := strings.Contains(host, ":")
+		port := query.Get("port")
+		if port != "" && !hasPort {
+			host = fmt.Sprintf("%s:%s", host, port)
+		} else {
+			http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+			return
+		}
+		result, err = monitor.NewRawHealthChecker(types.NewURL(&url.URL{
+			Scheme: scheme,
+			Host:   host,
+		}), defaultHealthConfig).CheckHealth()
+	}
+
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadGateway)
+		return
+	}
+
+	gphttp.RespondJSON(w, r, result)
+}

agent/pkg/handler/check_health_test.go

@@ -0,0 +1,216 @@
+package handler_test
+
+import (
+	"encoding/json"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strconv"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/yusing/go-proxy/agent/pkg/agent"
+	"github.com/yusing/go-proxy/agent/pkg/handler"
+	"github.com/yusing/go-proxy/internal/watcher/health"
+)
+
+func TestCheckHealthHTTP(t *testing.T) {
+	tests := []struct {
+		name            string
+		setupServer     func() *httptest.Server
+		queryParams     map[string]string
+		expectedStatus  int
+		expectedHealthy bool
+	}{
+		{
+			name: "Valid",
+			setupServer: func() *httptest.Server {
+				return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					w.WriteHeader(http.StatusOK)
+				}))
+			},
+			queryParams: map[string]string{
+				"scheme": "http",
+				"host":   "localhost",
+				"path":   "/",
+			},
+			expectedStatus:  http.StatusOK,
+			expectedHealthy: true,
+		},
+		{
+			name:        "InvalidQuery",
+			setupServer: nil,
+			queryParams: map[string]string{
+				"scheme": "http",
+			},
+			expectedStatus: http.StatusBadRequest,
+		},
+		{
+			name:        "ConnectionError",
+			setupServer: nil,
+			queryParams: map[string]string{
+				"scheme": "http",
+				"host":   "localhost:12345",
+			},
+			expectedStatus:  http.StatusOK,
+			expectedHealthy: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var server *httptest.Server
+			if tt.setupServer != nil {
+				server = tt.setupServer()
+				defer server.Close()
+				u, _ := url.Parse(server.URL)
+				tt.queryParams["scheme"] = u.Scheme
+				tt.queryParams["host"] = u.Host
+				tt.queryParams["path"] = u.Path
+			}
+
+			recorder := httptest.NewRecorder()
+			query := url.Values{}
+			for key, value := range tt.queryParams {
+				query.Set(key, value)
+			}
+			request := httptest.NewRequest(http.MethodGet, agent.APIEndpointBase+agent.EndpointHealth+"?"+query.Encode(), nil)
+			handler.CheckHealth(recorder, request)
+
+			require.Equal(t, recorder.Code, tt.expectedStatus)
+
+			if tt.expectedStatus == http.StatusOK {
+				var result health.HealthCheckResult
+				require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &result))
+				require.Equal(t, result.Healthy, tt.expectedHealthy)
+			}
+		})
+	}
+}
+
+func TestCheckHealthFileServer(t *testing.T) {
+	tests := []struct {
+		name            string
+		path            string
+		expectedStatus  int
+		expectedHealthy bool
+		expectedDetail  string
+	}{
+		{
+			name:            "ValidPath",
+			path:            t.TempDir(),
+			expectedStatus:  http.StatusOK,
+			expectedHealthy: true,
+			expectedDetail:  "",
+		},
+		{
+			name:            "InvalidPath",
+			path:            "/invalid",
+			expectedStatus:  http.StatusOK,
+			expectedHealthy: false,
+			expectedDetail:  "stat /invalid: no such file or directory",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			query := url.Values{}
+			query.Set("scheme", "fileserver")
+			query.Set("path", tt.path)
+
+			recorder := httptest.NewRecorder()
+			request := httptest.NewRequest(http.MethodGet, agent.APIEndpointBase+agent.EndpointHealth+"?"+query.Encode(), nil)
+			handler.CheckHealth(recorder, request)
+
+			require.Equal(t, recorder.Code, tt.expectedStatus)
+
+			var result health.HealthCheckResult
+			require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &result))
+			require.Equal(t, result.Healthy, tt.expectedHealthy)
+			require.Equal(t, result.Detail, tt.expectedDetail)
+		})
+	}
+}
+
+func TestCheckHealthTCPUDP(t *testing.T) {
+	tcp, err := net.Listen("tcp", "localhost:0")
+	require.NoError(t, err)
+	go func() {
+		conn, err := tcp.Accept()
+		require.NoError(t, err)
+		conn.Close()
+	}()
+
+	udp, err := net.ListenPacket("udp", "localhost:0")
+	require.NoError(t, err)
+	go func() {
+		buf := make([]byte, 1024)
+		n, addr, err := udp.ReadFrom(buf)
+		require.NoError(t, err)
+		require.Equal(t, string(buf[:n]), "ping")
+		_, _ = udp.WriteTo([]byte("pong"), addr)
+		udp.Close()
+	}()
+
+	tests := []struct {
+		name            string
+		scheme          string
+		host            string
+		port            int
+		expectedStatus  int
+		expectedHealthy bool
+	}{
+		{
+			name:            "ValidTCP",
+			scheme:          "tcp",
+			host:            "localhost",
+			port:            tcp.Addr().(*net.TCPAddr).Port,
+			expectedStatus:  http.StatusOK,
+			expectedHealthy: true,
+		},
+		{
+			name:            "InvalidHost",
+			scheme:          "tcp",
+			host:            "invalid",
+			port:            8080,
+			expectedStatus:  http.StatusOK,
+			expectedHealthy: false,
+		},
+		{
+			name:            "ValidUDP",
+			scheme:          "udp",
+			host:            "localhost",
+			port:            udp.LocalAddr().(*net.UDPAddr).Port,
+			expectedStatus:  http.StatusOK,
+			expectedHealthy: true,
+		},
+		{
+			name:            "InvalidHost",
+			scheme:          "udp",
+			host:            "invalid",
+			port:            8080,
+			expectedStatus:  http.StatusOK,
+			expectedHealthy: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			query := url.Values{}
+			query.Set("scheme", tt.scheme)
+			query.Set("host", tt.host)
+			query.Set("port", strconv.Itoa(tt.port))
+
+			recorder := httptest.NewRecorder()
+			request := httptest.NewRequest(http.MethodGet, agent.APIEndpointBase+agent.EndpointHealth+"?"+query.Encode(), nil)
+			handler.CheckHealth(recorder, request)
+
+			require.Equal(t, recorder.Code, tt.expectedStatus)
+
+			var result health.HealthCheckResult
+			require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &result))
+			require.Equal(t, result.Healthy, tt.expectedHealthy)
+		})
+	}
+}

agent/pkg/handler/docker_socket.go

@@ -0,0 +1,31 @@
+package handler
+
+import (
+	"net/http"
+	"net/url"
+
+	"github.com/docker/docker/client"
+	"github.com/yusing/go-proxy/internal/common"
+	"github.com/yusing/go-proxy/internal/docker"
+	"github.com/yusing/go-proxy/internal/logging"
+	"github.com/yusing/go-proxy/internal/net/gphttp/reverseproxy"
+	"github.com/yusing/go-proxy/internal/net/types"
+)
+
+func serviceUnavailable(w http.ResponseWriter, r *http.Request) {
+	http.Error(w, "docker socket is not available", http.StatusServiceUnavailable)
+}
+
+func DockerSocketHandler() http.HandlerFunc {
+	dockerClient, err := docker.NewClient(common.DockerHostFromEnv)
+	if err != nil {
+		logging.Warn().Err(err).Msg("failed to connect to docker client")
+		return serviceUnavailable
+	}
+	rp := reverseproxy.NewReverseProxy("docker", types.NewURL(&url.URL{
+		Scheme: "http",
+		Host:   client.DummyHost,
+	}), dockerClient.HTTPClient().Transport)
+
+	return rp.ServeHTTP
+}

agent/pkg/handler/handler.go

@@ -0,0 +1,49 @@
+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/metrics/systeminfo"
+	"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 NewAgentHandler() 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.EndpointHealth, CheckHealth)
+	mux.HandleMethods("GET", agent.EndpointLogs, memlogger.HandlerFunc())
+	mux.HandleMethods("GET", agent.EndpointSystemInfo, systeminfo.Poller.ServeHTTP)
+	mux.ServeMux.HandleFunc("/", DockerSocketHandler())
+	return mux
+}

agent/pkg/handler/proxy_http.go

@@ -0,0 +1,63 @@
+package handler
+
+import (
+	"crypto/tls"
+	"net/http"
+	"net/url"
+	"strconv"
+	"time"
+
+	"github.com/yusing/go-proxy/agent/pkg/agent"
+	"github.com/yusing/go-proxy/agent/pkg/agentproxy"
+	"github.com/yusing/go-proxy/internal/logging"
+	"github.com/yusing/go-proxy/internal/net/gphttp"
+	"github.com/yusing/go-proxy/internal/net/gphttp/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
+	}
+
+	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.ResponseHeaderTimeout = time.Duration(responseHeaderTimeout) * time.Second
+	}
+
+	r.URL.Scheme = ""
+	r.URL.Host = ""
+	r.URL.Path = r.URL.Path[agent.HTTPProxyURLPrefixLen:] // strip the {API_BASE}/proxy/http prefix
+	r.RequestURI = r.URL.String()
+	r.URL.Host = host
+	r.URL.Scheme = scheme
+
+	logging.Debug().Msgf("proxy http request: %s %s", r.Method, r.URL.String())
+
+	rp := reverseproxy.NewReverseProxy("agent", types.NewURL(&url.URL{
+		Scheme: scheme,
+		Host:   host,
+	}), transport)
+	rp.ServeHTTP(w, r)
+}