diff --git a/go.mod b/go.mod index 9e73c501..066f0ae8 100644 --- a/go.mod +++ b/go.mod @@ -11,6 +11,8 @@ require ( github.com/gin-gonic/gin v1.10.0 github.com/golang-jwt/jwt/v5 v5.2.2 github.com/google/uuid v1.6.0 + github.com/prometheus/client_golang v1.23.2 + github.com/prometheus/client_model v0.6.2 github.com/redis/go-redis/v9 v9.17.1 github.com/stretchr/testify v1.11.1 github.com/valkey-io/valkey-go v1.0.69 @@ -75,8 +77,6 @@ require ( github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/pelletier/go-toml/v2 v2.2.2 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect - github.com/prometheus/client_golang v1.23.2 // indirect - github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.67.1 // indirect github.com/prometheus/procfs v0.17.0 // indirect github.com/spf13/pflag v1.0.10 // indirect diff --git a/pkg/picod/execute.go b/pkg/picod/execute.go index d493f5fd..2319a0e2 100644 --- a/pkg/picod/execute.go +++ b/pkg/picod/execute.go @@ -53,8 +53,12 @@ type ExecuteResponse struct { // ExecuteHandler handles command execution requests func (s *Server) ExecuteHandler(c *gin.Context) { + s.metrics.ActiveExecutions.Inc() + defer s.metrics.ActiveExecutions.Dec() + var req ExecuteRequest if err := c.ShouldBindJSON(&req); err != nil { + s.metrics.ExecuteRequestsTotal.WithLabelValues("invalid").Inc() c.JSON(http.StatusBadRequest, gin.H{ "error": err.Error(), "code": http.StatusBadRequest, @@ -63,6 +67,7 @@ func (s *Server) ExecuteHandler(c *gin.Context) { } if len(req.Command) == 0 { + s.metrics.ExecuteRequestsTotal.WithLabelValues("invalid").Inc() c.JSON(http.StatusBadRequest, gin.H{ "error": "command cannot be empty", "code": http.StatusBadRequest, @@ -76,6 +81,7 @@ func (s *Server) ExecuteHandler(c *gin.Context) { var err error timeoutDuration, err = time.ParseDuration(req.Timeout) if err != nil { + s.metrics.ExecuteRequestsTotal.WithLabelValues("invalid").Inc() c.JSON(http.StatusBadRequest, gin.H{ "error": fmt.Sprintf("Invalid timeout format: %v", err), "code": http.StatusBadRequest, @@ -92,28 +98,11 @@ func (s *Server) ExecuteHandler(c *gin.Context) { // Use the first element as the command and the rest as arguments cmd := exec.CommandContext(ctx, req.Command[0], req.Command[1:]...) //nolint:gosec // This is an agent designed to execute arbitrary commands - // Default working directory to workspace; override if the request specifies one. - cmd.Dir = s.workspaceDir - if req.WorkingDir != "" { - safeWorkingDir, err := s.sanitizePath(req.WorkingDir) - if err != nil { - c.JSON(http.StatusBadRequest, gin.H{ - "error": fmt.Sprintf("Invalid working directory: %v", err), - "code": http.StatusBadRequest, - }) - return - } - if _, statErr := os.Stat(safeWorkingDir); os.IsNotExist(statErr) { - if err := s.mkdirSafe(safeWorkingDir); err != nil { - c.JSON(http.StatusInternalServerError, gin.H{ - "error": fmt.Sprintf("Failed to create working directory: %v", err), - "code": http.StatusInternalServerError, - }) - return - } - } - cmd.Dir = safeWorkingDir + workingDir, ok := s.prepareWorkingDir(c, req.WorkingDir) + if !ok { + return } + cmd.Dir = workingDir // Set environment variables if len(req.Env) > 0 { @@ -134,13 +123,21 @@ func (s *Server) ExecuteHandler(c *gin.Context) { endTime := time.Now() var exitCode int + var outcomeStatus string if errors.Is(ctx.Err(), context.DeadlineExceeded) { exitCode = TimeoutExitCode + outcomeStatus = "timeout" stderr.WriteString(fmt.Sprintf("Command timed out after %.0f seconds", timeoutDuration.Seconds())) } else if cmd.ProcessState != nil { exitCode = cmd.ProcessState.ExitCode() + if exitCode == 0 { + outcomeStatus = "success" + } else { + outcomeStatus = "error" + } } else { exitCode = 1 + outcomeStatus = "error" if stderr.Len() > 0 { stderr.WriteString("\n") } @@ -150,6 +147,8 @@ func (s *Server) ExecuteHandler(c *gin.Context) { } } + s.metrics.ExecuteRequestsTotal.WithLabelValues(outcomeStatus).Inc() + c.JSON(http.StatusOK, ExecuteResponse{ Stdout: stdout.String(), Stderr: stderr.String(), @@ -159,3 +158,29 @@ func (s *Server) ExecuteHandler(c *gin.Context) { EndTime: endTime, }) } + +func (s *Server) prepareWorkingDir(c *gin.Context, reqWorkingDir string) (string, bool) { + if reqWorkingDir == "" { + return s.workspaceDir, true + } + safeWorkingDir, err := s.sanitizePath(reqWorkingDir) + if err != nil { + s.metrics.ExecuteRequestsTotal.WithLabelValues("invalid").Inc() + c.JSON(http.StatusBadRequest, gin.H{ + "error": fmt.Sprintf("Invalid working directory: %v", err), + "code": http.StatusBadRequest, + }) + return "", false + } + if _, statErr := os.Stat(safeWorkingDir); os.IsNotExist(statErr) { + if err := s.mkdirSafe(safeWorkingDir); err != nil { + s.metrics.ExecuteRequestsTotal.WithLabelValues("error").Inc() + c.JSON(http.StatusInternalServerError, gin.H{ + "error": fmt.Sprintf("Failed to create working directory: %v", err), + "code": http.StatusInternalServerError, + }) + return "", false + } + } + return safeWorkingDir, true +} diff --git a/pkg/picod/metrics.go b/pkg/picod/metrics.go new file mode 100644 index 00000000..3ca0505c --- /dev/null +++ b/pkg/picod/metrics.go @@ -0,0 +1,139 @@ +/* +Copyright The Volcano Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package picod + +import ( + "net/http" + "strconv" + "time" + + "github.com/gin-gonic/gin" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +const methodUnknown = "unknown" + +// requestDurationBuckets covers PicoD's full latency range: sub-millisecond +// for fast file and health-style paths through the 60-second default command +// timeout used by ExecuteHandler. +var requestDurationBuckets = []float64{ + 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, + 1, 2.5, 5, 10, 20, 30, 45, 60, +} + +// normalizeMethod maps non-standard HTTP methods to "unknown" to prevent +// unbounded label cardinality in Prometheus time series. +func normalizeMethod(method string) string { + switch method { + case http.MethodGet, http.MethodPost, http.MethodPut, http.MethodDelete, + http.MethodPatch, http.MethodHead, http.MethodOptions, + http.MethodConnect, http.MethodTrace: + return method + default: + return methodUnknown + } +} + +// Metrics holds the Prometheus collectors for PicoD. +type Metrics struct { + Registry *prometheus.Registry + ActiveExecutions prometheus.Gauge + ExecuteRequestsTotal *prometheus.CounterVec + HTTPRequestsTotal *prometheus.CounterVec + HTTPRequestDuration *prometheus.HistogramVec +} + +// NewMetrics creates and registers metrics collectors with a private registry. +func NewMetrics() *Metrics { + reg := prometheus.NewRegistry() + + activeExecutions := prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "picod_active_executions", + Help: "Number of execute handler invocations currently in flight (including validation).", + }) + + executeRequestsTotal := prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "picod_execute_requests_total", + Help: "Total number of execute requests, partitioned by status (success, error, timeout, invalid).", + }, + []string{"status"}, + ) + + httpRequestsTotal := prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "picod_http_requests_total", + Help: "Total number of HTTP requests processed, partitioned by method, path, and status code.", + }, + []string{"method", "path", "status_code"}, + ) + + httpRequestDuration := prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "picod_http_request_duration_seconds", + Help: "Latency of HTTP requests, partitioned by method and path.", + Buckets: requestDurationBuckets, + }, + []string{"method", "path"}, + ) + + reg.MustRegister(activeExecutions) + reg.MustRegister(executeRequestsTotal) + reg.MustRegister(httpRequestsTotal) + reg.MustRegister(httpRequestDuration) + + return &Metrics{ + Registry: reg, + ActiveExecutions: activeExecutions, + ExecuteRequestsTotal: executeRequestsTotal, + HTTPRequestsTotal: httpRequestsTotal, + HTTPRequestDuration: httpRequestDuration, + } +} + +// Handler returns an HTTP handler for the metrics registry. +func (m *Metrics) Handler() gin.HandlerFunc { + h := promhttp.HandlerFor(m.Registry, promhttp.HandlerOpts{}) + return func(c *gin.Context) { + h.ServeHTTP(c.Writer, c.Request) + } +} + +// Middleware returns a Gin middleware that records HTTP request metrics. +func (m *Metrics) Middleware() gin.HandlerFunc { + return func(c *gin.Context) { + path := c.FullPath() + if path == "" { + path = "unmatched" + } + // Do not record requests to /metrics or /health to avoid noise + if path == "/metrics" || path == "/health" { + c.Next() + return + } + + start := time.Now() + c.Next() + duration := time.Since(start).Seconds() + + status := strconv.Itoa(c.Writer.Status()) + method := normalizeMethod(c.Request.Method) + m.HTTPRequestsTotal.WithLabelValues(method, path, status).Inc() + m.HTTPRequestDuration.WithLabelValues(method, path).Observe(duration) + } +} diff --git a/pkg/picod/metrics_test.go b/pkg/picod/metrics_test.go new file mode 100644 index 00000000..b24b9d16 --- /dev/null +++ b/pkg/picod/metrics_test.go @@ -0,0 +1,310 @@ +/* +Copyright The Volcano Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package picod + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "testing" + "time" + + "github.com/golang-jwt/jwt/v5" + dto "github.com/prometheus/client_model/go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + metricHTTPRequestsTotal = "picod_http_requests_total" + labelPath = "path" + labelMethod = "method" + executeAPIPath = "/api/execute" +) + +func TestMetrics_Exposition(t *testing.T) { + routerPriv, routerPubStr := generateRSAKeys(t) + server, ts, tmpDir := setupTestServer(t, routerPubStr) + defer os.RemoveAll(tmpDir) + defer ts.Close() + defer os.Unsetenv(PublicKeyEnvVar) + + client := ts.Client() + + // 1. Check /metrics endpoint is reachable and returns HTTP 200 + resp, err := client.Get(ts.URL + "/metrics") + require.NoError(t, err) + assert.Equal(t, http.StatusOK, resp.StatusCode) + resp.Body.Close() + + // 2. Perform an execution request + execReq := ExecuteRequest{ + Command: []string{"echo", "test-metrics"}, + } + bodyBytes, err := json.Marshal(execReq) + require.NoError(t, err) + + claims := jwt.MapClaims{ + "iat": time.Now().Unix(), + "exp": time.Now().Add(time.Hour * 6).Unix(), + } + token := createToken(t, routerPriv, claims) + + req, err := http.NewRequest(http.MethodPost, ts.URL+executeAPIPath, bytes.NewBuffer(bodyBytes)) + require.NoError(t, err) + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + + resp, err = client.Do(req) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, resp.StatusCode) + resp.Body.Close() + + // 3. Programmatically verify the metrics from the registry + metricFamilies, err := server.metrics.Registry.Gather() + require.NoError(t, err) + + var foundActiveExecutions, foundExecuteRequests, foundHTTPRequests, foundHTTPRequestDuration bool + + for _, mf := range metricFamilies { + switch *mf.Name { + case "picod_active_executions": + foundActiveExecutions = true + verifyActiveExecutions(t, mf) + + case "picod_execute_requests_total": + foundExecuteRequests = true + verifyExecuteRequests(t, mf) + + case metricHTTPRequestsTotal: + foundHTTPRequests = true + verifyHTTPRequests(t, mf) + + case "picod_http_request_duration_seconds": + foundHTTPRequestDuration = true + verifyHTTPRequestDuration(t, mf) + } + } + + assert.True(t, foundActiveExecutions, "picod_active_executions should be registered") + assert.True(t, foundExecuteRequests, "picod_execute_requests_total should be registered") + assert.True(t, foundHTTPRequests, "picod_http_requests_total should be registered") + assert.True(t, foundHTTPRequestDuration, "picod_http_request_duration_seconds should be registered") +} + +func verifyActiveExecutions(t *testing.T, mf *dto.MetricFamily) { + require.Len(t, mf.Metric, 1) + assert.Equal(t, 0.0, *mf.Metric[0].Gauge.Value) +} + +func verifyExecuteRequests(t *testing.T, mf *dto.MetricFamily) { + require.Len(t, mf.Metric, 1) + assert.Equal(t, 1.0, *mf.Metric[0].Counter.Value) + require.Len(t, mf.Metric[0].Label, 1) + assert.Equal(t, "status", *mf.Metric[0].Label[0].Name) + assert.Equal(t, "success", *mf.Metric[0].Label[0].Value) +} + +func verifyHTTPRequests(t *testing.T, mf *dto.MetricFamily) { + var foundExecute bool + for _, m := range mf.Metric { + var path, method, status string + for _, label := range m.Label { + switch *label.Name { + case labelPath: + path = *label.Value + case labelMethod: + method = *label.Value + case "status_code": + status = *label.Value + } + } + if path == executeAPIPath && method == http.MethodPost && status == "200" { + foundExecute = true + assert.Equal(t, 1.0, *m.Counter.Value) + } + } + assert.True(t, foundExecute, "should record POST /api/execute 200 metric") +} + +func verifyHTTPRequestDuration(t *testing.T, mf *dto.MetricFamily) { + var foundExecute bool + for _, m := range mf.Metric { + var path, method string + for _, label := range m.Label { + switch *label.Name { + case labelPath: + path = *label.Value + case labelMethod: + method = *label.Value + } + } + if path == executeAPIPath && method == http.MethodPost { + foundExecute = true + assert.Greater(t, *m.Histogram.SampleCount, uint64(0)) + assert.Greater(t, *m.Histogram.SampleSum, 0.0) + // Verify the extended bucket set was applied: all 15 explicit + // buckets must be present and the uppermost must cover the + // 60-second default command timeout. + require.Equal(t, len(requestDurationBuckets), len(m.Histogram.Bucket)) + lastBucket := m.Histogram.Bucket[len(m.Histogram.Bucket)-1] + assert.Equal(t, 60.0, *lastBucket.UpperBound) + } + } + assert.True(t, foundExecute, "should record duration for POST /api/execute") +} + +func TestMetrics_OversizedRequest(t *testing.T) { + _, routerPubStr := generateRSAKeys(t) + server, ts, tmpDir := setupTestServer(t, routerPubStr) + defer os.RemoveAll(tmpDir) + defer ts.Close() + defer os.Unsetenv(PublicKeyEnvVar) + + // Use ServeHTTP directly to set ContentLength without HTTP client validation. + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodPost, executeAPIPath, nil) + r.ContentLength = int64(MaxBodySize) + 1 + r.Header.Set("Content-Type", "application/json") + server.engine.ServeHTTP(w, r) + assert.Equal(t, http.StatusRequestEntityTooLarge, w.Code) + + metricFamilies, err := server.metrics.Registry.Gather() + require.NoError(t, err) + + var found413 bool + for _, mf := range metricFamilies { + if *mf.Name != metricHTTPRequestsTotal { + continue + } + for _, m := range mf.Metric { + var path, method, status string + for _, label := range m.Label { + switch *label.Name { + case labelPath: + path = *label.Value + case labelMethod: + method = *label.Value + case "status_code": + status = *label.Value + } + } + if path == executeAPIPath && method == http.MethodPost && status == "413" { + found413 = true + assert.Equal(t, 1.0, *m.Counter.Value) + } + } + } + assert.True(t, found413, "oversized request should be recorded with POST /api/execute status_code=413") +} + +func TestMetrics_UnmatchedRoute(t *testing.T) { + _, routerPubStr := generateRSAKeys(t) + server, ts, tmpDir := setupTestServer(t, routerPubStr) + defer os.RemoveAll(tmpDir) + defer ts.Close() + defer os.Unsetenv(PublicKeyEnvVar) + + client := ts.Client() + + resp, err := client.Get(ts.URL + "/does-not-exist") + require.NoError(t, err) + assert.Equal(t, http.StatusNotFound, resp.StatusCode) + resp.Body.Close() + + metricFamilies, err := server.metrics.Registry.Gather() + require.NoError(t, err) + + var foundUnmatched bool + for _, mf := range metricFamilies { + if *mf.Name != metricHTTPRequestsTotal { + continue + } + for _, m := range mf.Metric { + for _, label := range m.Label { + if *label.Name == labelPath && *label.Value == "unmatched" { + foundUnmatched = true + } + } + } + } + assert.True(t, foundUnmatched, "unmatched routes must use path label 'unmatched', not the raw URL path") +} + +func TestNormalizeMethod(t *testing.T) { + standard := []string{ + http.MethodGet, http.MethodPost, http.MethodPut, http.MethodDelete, + http.MethodPatch, http.MethodHead, http.MethodOptions, + http.MethodConnect, http.MethodTrace, + } + for _, m := range standard { + assert.Equal(t, m, normalizeMethod(m), "standard method must be preserved: %s", m) + } + + custom := []string{"FOO", "BAR", "RANDOM123", "PROPFIND", "MKCOL"} + for _, m := range custom { + assert.Equal(t, methodUnknown, normalizeMethod(m), "non-standard method must become %q: %s", methodUnknown, m) + } +} + +func TestMetrics_UnknownMethod(t *testing.T) { + _, routerPubStr := generateRSAKeys(t) + server, ts, tmpDir := setupTestServer(t, routerPubStr) + defer os.RemoveAll(tmpDir) + defer ts.Close() + defer os.Unsetenv(PublicKeyEnvVar) + + // Send three requests each with a distinct non-standard HTTP method. + customMethods := []string{"FOO", "BAR", "RANDOM123"} + for _, m := range customMethods { + w := httptest.NewRecorder() + r := httptest.NewRequest(m, executeAPIPath, nil) + server.engine.ServeHTTP(w, r) + } + + metricFamilies, err := server.metrics.Registry.Gather() + require.NoError(t, err) + + var unknownSeries int + var unknownTotal float64 + var customLabelFound bool + for _, mf := range metricFamilies { + if *mf.Name != metricHTTPRequestsTotal { + continue + } + for _, m := range mf.Metric { + for _, label := range m.Label { + if *label.Name != labelMethod { + continue + } + switch *label.Value { + case methodUnknown: + unknownSeries++ + unknownTotal += *m.Counter.Value + case "FOO", "BAR", "RANDOM123": + customLabelFound = true + } + } + } + } + + assert.Equal(t, 1, unknownSeries, "all non-standard methods must collapse into a single %q series", methodUnknown) + assert.Equal(t, float64(len(customMethods)), unknownTotal, "counter must reflect all requests in the single series") + assert.False(t, customLabelFound, "non-standard method names must not appear as metric label values") +} diff --git a/pkg/picod/server.go b/pkg/picod/server.go index 2bae2ff4..b0d69fc0 100644 --- a/pkg/picod/server.go +++ b/pkg/picod/server.go @@ -46,6 +46,7 @@ type Server struct { authManager *AuthManager startTime time.Time workspaceDir string + metrics *Metrics } // NewServer creates a new PicoD server instance @@ -54,6 +55,7 @@ func NewServer(config Config) *Server { config: config, startTime: time.Now(), authManager: NewAuthManager(), + metrics: NewMetrics(), } // Initialize workspace directory @@ -77,11 +79,12 @@ func NewServer(config Config) *Server { engine := gin.New() // Global middleware - engine.Use(gin.Logger()) // Request logging + engine.Use(gin.Logger()) // Request logging + engine.Use(s.metrics.Middleware()) engine.Use(gin.Recovery()) // Crash recovery engine.Use(maxBodySizeMiddleware()) engine.MaxMultipartMemory = MaxBodySize - engine.Use(gzip.Gzip(gzip.BestSpeed, gzip.WithExcludedPaths([]string{"/health"}))) // Response compression + engine.Use(gzip.Gzip(gzip.BestSpeed, gzip.WithExcludedPaths([]string{"/health", "/metrics"}))) // Response compression // Load public key from environment variable (required for JWT auth) if err := s.authManager.LoadPublicKeyFromEnv(); err != nil { @@ -101,6 +104,9 @@ func NewServer(config Config) *Server { // Health check (no authentication required) engine.GET("/health", s.HealthCheckHandler) + // Metrics endpoint (no authentication required) + engine.GET("/metrics", s.metrics.Handler()) + s.engine = engine return s }