[elastic-agent] proxy requests to subprocesses to their metrics endpoints

x-pack/elastic-agent/CHANGELOG.next.asciidoc

@@ -13,6 +13,7 @@
 - Remove the `--kibana-url` from `install` and `enroll` command. {pull}25529[25529]
 - Default to port 80 and 443 for Kibana and Fleet Server connections. {pull}25723[25723]
 - Remove deprecated/undocumented IncludeCreatorMetadata setting from kubernetes metadata config options {pull}28006[28006]
+- The `/processes/<subprocess>` endpoint proxies to the subprocess's monitoring endpoint, instead of querying its `/stats` endpoint {pull}28165[28165]
 
 ==== Bugfixes
 - Fix rename *ConfigChange to *PolicyChange to align on changes in the UI. {pull}20779[20779]

x-pack/elastic-agent/pkg/core/monitoring/server/process.go

@@ -37,6 +37,7 @@ const (
 var (
 	// ErrProgramNotSupported returned when requesting metrics for not supported program.
 	ErrProgramNotSupported = errors.New("specified program is not supported")
+	errPathNotFound        = errors.New("endpoint not found")
 	invalidChars           = map[rune]struct{}{
 		'"':  {},
 		'<':  {},
@@ -67,8 +68,13 @@ func processHandler(statsHandler func(http.ResponseWriter, *http.Request) error)
 			// proxy stats for elastic agent process
 			return statsHandler(w, r)
 		}
+		beatsPath := vars["beatsPath"]
 
-		metricsBytes, statusCode, metricsErr := processMetrics(r.Context(), id)
+		endpoint, err := generateEndpoint(id)
+		if err != nil {
+			return err
+		}
+		metricsBytes, statusCode, metricsErr := processMetrics(r.Context(), endpoint, beatsPath)
 		if metricsErr != nil {
 			return metricsErr
 		}
@@ -82,23 +88,18 @@ func processHandler(statsHandler func(http.ResponseWriter, *http.Request) error)
 	}
 }
 
-func processMetrics(ctx context.Context, id string) ([]byte, int, error) {
-	detail, err := parseID(id)
-	if err != nil {
-		return nil, 0, err
-	}
-
-	endpoint := beats.MonitoringEndpoint(detail.spec, artifact.DefaultConfig().OS(), detail.output)
-	if !strings.HasPrefix(endpoint, httpPlusPrefix) && !strings.HasPrefix(endpoint, "http") {
-		// add prefix for npipe and unix
-		endpoint = httpPlusPrefix + endpoint
-	}
+var beatsPathAllowlist = map[string]struct{}{
+	"":      struct{}{},
+	"stats": struct{}{},
+	"state": struct{}{},
+}
 
-	if detail.isMonitoring {
-		endpoint += "_monitor"
+func processMetrics(ctx context.Context, endpoint, path string) ([]byte, int, error) {
+	if _, ok := beatsPathAllowlist[path]; !ok {
+		return nil, http.StatusNotFound, errPathNotFound
 	}
 
-	hostData, err := parse.ParseURL(endpoint, "http", "", "", "stats", "")
+	hostData, err := parse.ParseURL(endpoint, "http", "", "", path, "")
 	if err != nil {
 		return nil, 0, errorWithStatus(http.StatusInternalServerError, err)
 	}
@@ -145,6 +146,24 @@ func processMetrics(ctx context.Context, id string) ([]byte, int, error) {
 	return rb, resp.StatusCode, nil
 }
 
+func generateEndpoint(id string) (string, error) {
+	detail, err := parseID(id)
+	if err != nil {
+		return "", err
+	}
+
+	endpoint := beats.MonitoringEndpoint(detail.spec, artifact.DefaultConfig().OS(), detail.output)
+	if !strings.HasPrefix(endpoint, httpPlusPrefix) && !strings.HasPrefix(endpoint, "http") {
+		// add prefix for npipe and unix
+		endpoint = httpPlusPrefix + endpoint
+	}
+
+	if detail.isMonitoring {
+		endpoint += "_monitor"
+	}
+	return endpoint, nil
+}
+
 func writeResponse(w http.ResponseWriter, c interface{}) {
 	bytes, err := json.Marshal(c)
 	if err != nil {

x-pack/elastic-agent/pkg/core/monitoring/server/process_linux_test.go

@@ -0,0 +1,46 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+//go:build linux
+// +build linux
+
+package server
+
+import (
+	"context"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestProcessProxyRequest(t *testing.T) {
+	sock := "/tmp/elastic-agent-test.sock"
+	defer os.Remove(sock)
+
+	endpoint := "http+unix://" + sock
+	server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Write the path to the client so they can verify the request
+		// was correct
+		w.Write([]byte(r.URL.Path))
+	}))
+
+	// Mimic subprocesses and listen on a unix socket
+	l, err := net.Listen("unix", sock)
+	require.NoError(t, err)
+	server.Listener = l
+	server.Start()
+	defer server.Close()
+
+	for _, path := range []string{"stats", "", "state"} {
+		respBytes, _, err := processMetrics(context.Background(), endpoint, path)
+		require.NoError(t, err)
+		// Verify that the server saw the path we tried to request
+		assert.Equal(t, "/"+path, string(respBytes))
+	}
+}

x-pack/elastic-agent/pkg/core/monitoring/server/process_test.go

@@ -4,11 +4,14 @@
 package server
 
 import (
+	"context"
 	"errors"
 	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	"gotest.tools/assert"
 )
 
 func TestParseID(t *testing.T) {
@@ -87,3 +90,31 @@ func TestStatusErr(t *testing.T) {
 		})
 	}
 }
+
+func TestBeatsPathAllowlist(t *testing.T) {
+	ctx := context.Background()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte("OK"))
+	}))
+	defer server.Close()
+	endpoint := server.URL
+
+	tcs := []struct {
+		path       string
+		statusCode int
+	}{
+		{"", http.StatusOK},
+		{"stats", http.StatusOK},
+		{"state", http.StatusOK},
+		{"not-allowed", http.StatusNotFound},
+		{"nested/path", http.StatusNotFound},
+	}
+
+	for _, tc := range tcs {
+		_, respCode, err := processMetrics(ctx, endpoint, tc.path)
+		assert.Equal(t, respCode, tc.statusCode)
+		if tc.statusCode > 299 {
+			assert.Error(t, err, errPathNotFound.Error())
+		}
+	}
+}

x-pack/elastic-agent/pkg/core/monitoring/server/server.go

@@ -50,6 +50,8 @@ func exposeMetricsEndpoint(log *logger.Logger, config *common.Config, ns func(st
 	if enableProcessStats {
 		r.HandleFunc("/processes", processesHandler(routesFetchFn))
 		r.Handle("/processes/{processID}", createHandler(processHandler(statsHandler)))
+		r.Handle("/processes/{processID}/", createHandler(processHandler(statsHandler)))
+		r.Handle("/processes/{processID}/{beatsPath}", createHandler(processHandler(statsHandler)))
 	}
 
 	mux := http.NewServeMux()