From 0165b9695890b8e44a0e064fe1080c87b6d1526a Mon Sep 17 00:00:00 2001 From: Luan Dang Date: Wed, 9 Sep 2026 22:32:00 -0500 Subject: [PATCH 1/3] api: add metrics-only listener --- AUTHORS | 1 + api/api_test.go | 7 + api/router.go | 4 +- cmd/api_serve.go | 252 ++++++++++++++++- cmd/api_serve_test.go | 422 +++++++++++++++++++++++++++++ completion.d/_aptly | 1 + completion.d/aptly | 2 +- man/aptly.1 | 9 +- system/t12_api/metrics_listener.py | 155 +++++++++++ 9 files changed, 839 insertions(+), 14 deletions(-) create mode 100644 cmd/api_serve_test.go create mode 100644 system/t12_api/metrics_listener.py diff --git a/AUTHORS b/AUTHORS index ea69b2987..816ab25ac 100644 --- a/AUTHORS +++ b/AUTHORS @@ -84,3 +84,4 @@ List of contributors, in chronological order: * Zhang Xiao (https://github.com/xzhang1) * Tom Nguyen (https://github.com/lecafard) * Philip Cramer (https://github.com/PhilipCramer) +* Luan Dang (https://github.com/luanmdang) diff --git a/api/api_test.go b/api/api_test.go index b6a5b11f1..986d131c9 100644 --- a/api/api_test.go +++ b/api/api_test.go @@ -158,6 +158,13 @@ func (s *APISuite) TestGetMetrics(c *C) { c.Check(b, Matches, ".*aptly_build_info.*version=\"testVersion\".*") } +func (s *APISuite) TestHeadMetrics(c *C) { + response, err := s.HTTPRequest("HEAD", "/api/metrics", nil) + c.Assert(err, IsNil) + c.Check(response.Code, Equals, 200) + c.Check(strings.Contains(response.Body.String(), "# TYPE aptly_build_info gauge"), Equals, true) +} + func (s *APISuite) TestRepoCreate(c *C) { body, err := json.Marshal(gin.H{ "Name": "dummy", diff --git a/api/router.go b/api/router.go index 1818b2a12..fdf6f268a 100644 --- a/api/router.go +++ b/api/router.go @@ -122,7 +122,9 @@ func Router(c *ctx.AptlyContext) http.Handler { { if c.Config().EnableMetricsEndpoint { - api.GET("/metrics", apiMetricsGet()) + metricsHandler := apiMetricsGet() + api.GET("/metrics", metricsHandler) + api.HEAD("/metrics", metricsHandler) } api.GET("/version", apiVersion) api.GET("/storage", apiDiskFree) diff --git a/cmd/api_serve.go b/cmd/api_serve.go index 17a061b3e..457c817c3 100644 --- a/cmd/api_serve.go +++ b/cmd/api_serve.go @@ -9,7 +9,9 @@ import ( "net/url" "os" "os/signal" + "sync" "syscall" + "time" "github.com/aptly-dev/aptly/api" "github.com/aptly-dev/aptly/systemd/activation" @@ -18,6 +20,214 @@ import ( "github.com/smira/flag" ) +const ( + metricsReadHeaderTimeout = 5 * time.Second + httpShutdownTimeout = 30 * time.Second +) + +type apiHTTPServer struct { + name string + server *http.Server + listener net.Listener +} + +type apiServeResult struct { + name string + err error +} + +func metricsOnlyHandler(apiHandler http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.EscapedPath() != "/metrics" { + http.NotFound(w, r) + return + } + + if r.Method != http.MethodGet && r.Method != http.MethodHead { + w.Header().Set("Allow", "GET, HEAD") + http.Error(w, http.StatusText(http.StatusMethodNotAllowed), http.StatusMethodNotAllowed) + return + } + + request := r.Clone(r.Context()) + requestURL := *r.URL + requestURL.Path = "/api/metrics" + requestURL.RawPath = "" + request.URL = &requestURL + request.RequestURI = requestURL.RequestURI() + apiHandler.ServeHTTP(w, request) + }) +} + +func validateMetricsListener(address string, metricsEnabled bool) error { + if address != "" && !metricsEnabled { + return errors.New("-metrics-listen requires enableMetricsEndpoint to be true") + } + return nil +} + +func selectActivatedAPIListener(listeners []net.Listener) (net.Listener, error) { + switch len(listeners) { + case 0: + return nil, nil + case 1: + if listeners[0] == nil { + return nil, errors.New("systemd file descriptor is not a supported network listener") + } + return listeners[0], nil + default: + return nil, fmt.Errorf("got %d listeners from systemd; only one API listener is supported", len(listeners)) + } +} + +func listenForAPI(address string) (net.Listener, error) { + listenURL, err := url.Parse(address) + if err == nil && listenURL.Scheme == "unix" { + file := listenURL.Path + _ = os.Remove(file) + + listener, err := net.Listen("unix", file) + if err != nil { + return nil, fmt.Errorf("failed to listen on API Unix socket %q: %w", file, err) + } + return listener, nil + } + + listener, err := net.Listen("tcp", address) + if err != nil { + return nil, fmt.Errorf("failed to listen on API address %q: %w", address, err) + } + return listener, nil +} + +func shutdownHTTPServers(servers []apiHTTPServer, timeout time.Duration) { + shutdownContext, cancel := stdcontext.WithTimeout(stdcontext.Background(), timeout) + defer cancel() + + failed := make(chan *http.Server, len(servers)) + var waitGroup sync.WaitGroup + for _, httpServer := range servers { + waitGroup.Add(1) + go func() { + defer waitGroup.Done() + if err := httpServer.server.Shutdown(shutdownContext); err != nil { + failed <- httpServer.server + } + }() + } + waitGroup.Wait() + close(failed) + + for server := range failed { + _ = server.Close() + } +} + +func serveHTTPServers(servers []apiHTTPServer, sigchan <-chan os.Signal, restoreSignals func(), waitForTasks func(), shutdownTimeout time.Duration) error { + results := make(chan apiServeResult, len(servers)) + for _, httpServer := range servers { + go func() { + results <- apiServeResult{ + name: httpServer.name, + err: httpServer.server.Serve(httpServer.listener), + } + }() + } + + serveResults := make([]apiServeResult, 0, len(servers)) + shutdownFromSignal := false + select { + case result := <-results: + serveResults = append(serveResults, result) + case <-sigchan: + shutdownFromSignal = true + } + + if restoreSignals != nil { + restoreSignals() + } + + if shutdownFromSignal { + fmt.Printf("\nShutdown signal received, stopping HTTP servers...\n") + } else { + fmt.Printf("\nHTTP server stopped unexpectedly, stopping sibling server...\n") + } + shutdownHTTPServers(servers, shutdownTimeout) + + for len(serveResults) < len(servers) { + serveResults = append(serveResults, <-results) + } + + fmt.Printf("Waiting for background tasks...\n") + waitForTasks() + + if shutdownFromSignal { + for _, result := range serveResults { + if result.err != nil && !errors.Is(result.err, http.ErrServerClosed) { + return fmt.Errorf("%s server failed during shutdown: %w", result.name, result.err) + } + } + return nil + } + + result := serveResults[0] + if result.err == nil { + return fmt.Errorf("%s server stopped unexpectedly", result.name) + } + return fmt.Errorf("%s server stopped unexpectedly: %w", result.name, result.err) +} + +func serveAPIWithMetrics(apiListener net.Listener, activated bool, apiAddress, metricsAddress string) error { + var err error + if apiListener == nil { + apiListener, err = listenForAPI(apiAddress) + if err != nil { + return err + } + } + defer func() { _ = apiListener.Close() }() + + metricsListener, err := net.Listen("tcp", metricsAddress) + if err != nil { + return fmt.Errorf("failed to listen on metrics address %q: %w", metricsAddress, err) + } + defer func() { _ = metricsListener.Close() }() + + apiHandler := api.Router(context) + servers := []apiHTTPServer{ + { + name: "API", + server: &http.Server{Handler: apiHandler}, + listener: apiListener, + }, + { + name: "metrics", + server: &http.Server{ + Handler: metricsOnlyHandler(apiHandler), + ReadHeaderTimeout: metricsReadHeaderTimeout, + }, + listener: metricsListener, + }, + } + + if activated { + fmt.Printf("\nTaking over API web server at: %s (press Ctrl+C to quit)...\n", apiListener.Addr().String()) + } else { + fmt.Printf("\nStarting API web server at: %s (press Ctrl+C to quit)...\n", apiListener.Addr().String()) + } + fmt.Printf("Starting metrics web server at: %s...\n", metricsListener.Addr().String()) + + sigchan := make(chan os.Signal, 1) + signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM) + defer signal.Stop(sigchan) + + restoreSignals := func() { + signal.Reset(syscall.SIGINT, syscall.SIGTERM) + } + + return serveHTTPServers(servers, sigchan, restoreSignals, context.TaskList().Wait, httpShutdownTimeout) +} + func aptlyAPIServe(cmd *commander.Command, args []string) error { var ( err error @@ -39,24 +249,37 @@ func aptlyAPIServe(cmd *commander.Command, args []string) error { return err } + metricsListen := context.Flags().Lookup("metrics-listen").Value.String() + if err = validateMetricsListener(metricsListen, context.Config().EnableMetricsEndpoint); err != nil { + return err + } + // Try to recycle systemd fds for listening listeners, err := activation.Listeners(true) - if len(listeners) > 1 { - panic("Got more than 1 listener from systemd. This is currently not supported!") - } - if err == nil && len(listeners) == 1 { - listener := listeners[0] - defer func() { _ = listener.Close() }() - fmt.Printf("\nTaking over web server at: %s (press Ctrl+C to quit)...\n", listener.Addr().String()) - err = http.Serve(listener, api.Router(context)) - if err != nil { - return fmt.Errorf("unable to serve: %s", err) + if err == nil { + listener, listenerErr := selectActivatedAPIListener(listeners) + if listenerErr != nil { + return listenerErr + } + if listener != nil && metricsListen != "" { + return serveAPIWithMetrics(listener, true, "", metricsListen) + } + if listener != nil { + defer func() { _ = listener.Close() }() + fmt.Printf("\nTaking over web server at: %s (press Ctrl+C to quit)...\n", listener.Addr().String()) + err = http.Serve(listener, api.Router(context)) + if err != nil { + return fmt.Errorf("unable to serve: %s", err) + } + return nil } - return nil } // If there are none: use the listen argument. listen := context.Flags().Lookup("listen").Value.String() + if metricsListen != "" { + return serveAPIWithMetrics(nil, false, listen, metricsListen) + } fmt.Printf("\nStarting web server at: %s (press Ctrl+C to quit)...\n", listen) server := http.Server{Handler: api.Router(context)} @@ -108,15 +331,22 @@ or Unix domain socket. When using a socket, Aptly will fully manage the socket file. This command also supports taking over from a systemd file descriptors to enable systemd socket activation. +An optional metrics-only TCP listener can be enabled with -metrics-listen. It +serves GET and HEAD requests at /metrics and requires enableMetricsEndpoint in +the aptly configuration. This listener uses plain HTTP; provide access control +and TLS through network policy or a proxy when required. + Example: $ aptly api serve -listen=:8080 $ aptly api serve -listen=unix:///tmp/aptly.sock + $ aptly api serve -listen=:8080 -metrics-listen=127.0.0.1:9090 `, Flag: *flag.NewFlagSet("aptly-serve", flag.ExitOnError), } cmd.Flag.String("listen", ":8080", "host:port for HTTP listening or unix://path to listen on a Unix domain socket") + cmd.Flag.String("metrics-listen", "", "host:port for optional metrics-only HTTP listening (requires enableMetricsEndpoint)") cmd.Flag.Bool("no-lock", false, "don't lock the database") return cmd diff --git a/cmd/api_serve_test.go b/cmd/api_serve_test.go new file mode 100644 index 000000000..d5220b9ae --- /dev/null +++ b/cmd/api_serve_test.go @@ -0,0 +1,422 @@ +package cmd + +import ( + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync" + "syscall" + "testing" + "time" + + check "gopkg.in/check.v1" +) + +func Test(t *testing.T) { + check.TestingT(t) +} + +func TestMetricsOnlyHandler(t *testing.T) { + tests := []struct { + name string + method string + target string + wantStatus int + wantForwarded bool + wantQuery string + wantRequestURI string + wantAllowMethods string + }{ + { + name: "get", + method: http.MethodGet, + target: "/metrics?_async=true&value=a%2Bb", + wantStatus: http.StatusNoContent, + wantForwarded: true, + wantQuery: "_async=true&value=a%2Bb", + wantRequestURI: "/api/metrics?_async=true&value=a%2Bb", + }, + { + name: "head", + method: http.MethodHead, + target: "/metrics", + wantStatus: http.StatusNoContent, + wantForwarded: true, + wantRequestURI: "/api/metrics", + }, + { + name: "api metrics path", + method: http.MethodGet, + target: "/api/metrics", + wantStatus: http.StatusNotFound, + }, + { + name: "trailing slash", + method: http.MethodGet, + target: "/metrics/", + wantStatus: http.StatusNotFound, + }, + { + name: "encoded alias", + method: http.MethodGet, + target: "/met%72ics", + wantStatus: http.StatusNotFound, + }, + { + name: "case variant", + method: http.MethodGet, + target: "/METRICS", + wantStatus: http.StatusNotFound, + }, + { + name: "unsupported method", + method: http.MethodPost, + target: "/metrics", + wantStatus: http.StatusMethodNotAllowed, + wantAllowMethods: "GET, HEAD", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + forwarded := false + var forwardedMethod, forwardedPath, forwardedQuery, forwardedRequestURI string + apiHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + forwarded = true + forwardedMethod = r.Method + forwardedPath = r.URL.Path + forwardedQuery = r.URL.RawQuery + forwardedRequestURI = r.RequestURI + w.WriteHeader(http.StatusNoContent) + }) + + request := httptest.NewRequest(test.method, test.target, nil) + originalPath := request.URL.Path + originalRawPath := request.URL.RawPath + originalRequestURI := request.RequestURI + response := httptest.NewRecorder() + + metricsOnlyHandler(apiHandler).ServeHTTP(response, request) + + if response.Code != test.wantStatus { + t.Fatalf("status = %d, want %d", response.Code, test.wantStatus) + } + if forwarded != test.wantForwarded { + t.Fatalf("forwarded = %t, want %t", forwarded, test.wantForwarded) + } + if got := response.Header().Get("Allow"); got != test.wantAllowMethods { + t.Errorf("Allow = %q, want %q", got, test.wantAllowMethods) + } + if request.URL.Path != originalPath || request.URL.RawPath != originalRawPath || request.RequestURI != originalRequestURI { + t.Errorf("original request was mutated") + } + + if test.wantForwarded { + if forwardedMethod != test.method { + t.Errorf("forwarded method = %q, want %q", forwardedMethod, test.method) + } + if forwardedPath != "/api/metrics" { + t.Errorf("forwarded path = %q, want /api/metrics", forwardedPath) + } + if forwardedQuery != test.wantQuery { + t.Errorf("forwarded query = %q, want %q", forwardedQuery, test.wantQuery) + } + if forwardedRequestURI != test.wantRequestURI { + t.Errorf("forwarded request URI = %q, want %q", forwardedRequestURI, test.wantRequestURI) + } + } + }) + } +} + +func TestValidateMetricsListener(t *testing.T) { + for _, test := range []struct { + name string + address string + metricsEnabled bool + wantError bool + }{ + {name: "disabled listener", metricsEnabled: false}, + {name: "existing API metrics only", metricsEnabled: true}, + {name: "enabled listener", address: "127.0.0.1:9090", metricsEnabled: true}, + {name: "listener requires metrics", address: "127.0.0.1:9090", metricsEnabled: false, wantError: true}, + } { + t.Run(test.name, func(t *testing.T) { + err := validateMetricsListener(test.address, test.metricsEnabled) + if (err != nil) != test.wantError { + t.Fatalf("error = %v, wantError %t", err, test.wantError) + } + }) + } +} + +func TestSelectActivatedAPIListener(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = listener.Close() }) + + selected, err := selectActivatedAPIListener(nil) + if err != nil || selected != nil { + t.Fatalf("no listeners: selected = %v, error = %v", selected, err) + } + + selected, err = selectActivatedAPIListener([]net.Listener{listener}) + if err != nil || selected != listener { + t.Fatalf("one listener: selected = %v, error = %v", selected, err) + } + + if _, err = selectActivatedAPIListener([]net.Listener{nil}); err == nil { + t.Fatal("nil listener did not return an error") + } + if _, err = selectActivatedAPIListener([]net.Listener{listener, listener}); err == nil { + t.Fatal("multiple listeners did not return an error") + } +} + +func TestServeAPIWithMetricsClosesAPIListenerOnMetricsBindFailure(t *testing.T) { + apiListener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = apiListener.Close() }) + apiAddress := apiListener.Addr().String() + + occupiedMetricsListener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + _ = apiListener.Close() + t.Fatal(err) + } + defer func() { _ = occupiedMetricsListener.Close() }() + + err = serveAPIWithMetrics(apiListener, false, "", occupiedMetricsListener.Addr().String()) + if err == nil || !strings.Contains(err.Error(), "failed to listen on metrics address") { + t.Fatalf("error = %v, want metrics bind failure", err) + } + + connection, dialErr := net.DialTimeout("tcp", apiAddress, 100*time.Millisecond) + if dialErr == nil { + _ = connection.Close() + t.Fatal("API listener still accepts connections after metrics bind failure") + } +} + +func TestShutdownHTTPServersForcesCloseAfterTimeout(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + + requestStarted := make(chan struct{}) + releaseRequest := make(chan struct{}) + var releaseOnce sync.Once + t.Cleanup(func() { + releaseOnce.Do(func() { close(releaseRequest) }) + }) + + server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + close(requestStarted) + <-releaseRequest + w.WriteHeader(http.StatusOK) + })} + serveResult := make(chan error, 1) + go func() { serveResult <- server.Serve(listener) }() + + clientResult := make(chan error, 1) + go func() { + response, requestErr := http.Get("http://" + listener.Addr().String()) + if response != nil { + _ = response.Body.Close() + } + clientResult <- requestErr + }() + + select { + case <-requestStarted: + case <-time.After(time.Second): + t.Fatal("request did not reach server") + } + + started := time.Now() + shutdownHTTPServers([]apiHTTPServer{{name: "test", server: server, listener: listener}}, 50*time.Millisecond) + if elapsed := time.Since(started); elapsed < 40*time.Millisecond { + t.Fatalf("shutdown closed an active request immediately after %s", elapsed) + } + if elapsed := time.Since(started); elapsed > time.Second { + t.Fatalf("forced shutdown took %s", elapsed) + } + + select { + case requestErr := <-clientResult: + if requestErr == nil { + t.Error("client request unexpectedly completed during forced shutdown") + } + case <-time.After(time.Second): + t.Fatal("client connection was not closed") + } + + select { + case serveErr := <-serveResult: + if serveErr != http.ErrServerClosed { + t.Errorf("Serve error = %v, want %v", serveErr, http.ErrServerClosed) + } + case <-time.After(time.Second): + t.Fatal("Serve did not return") + } + + releaseOnce.Do(func() { close(releaseRequest) }) +} + +func TestServeHTTPServersStopsSiblingOnFailure(t *testing.T) { + apiListener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + metricsListener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + _ = apiListener.Close() + t.Fatal(err) + } + + handler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNoContent) + }) + servers := []apiHTTPServer{ + {name: "API", server: &http.Server{Handler: handler}, listener: apiListener}, + {name: "metrics", server: &http.Server{Handler: handler}, listener: metricsListener}, + } + for _, server := range servers { + server := server + t.Cleanup(func() { + _ = server.server.Close() + _ = server.listener.Close() + }) + } + + sigchan := make(chan os.Signal) + signalsRestored := make(chan struct{}) + tasksWaited := make(chan struct{}) + result := make(chan error, 1) + go func() { + result <- serveHTTPServers( + servers, + sigchan, + func() { close(signalsRestored) }, + func() { close(tasksWaited) }, + time.Second, + ) + }() + + waitForHTTP(t, apiListener.Addr().String()) + waitForHTTP(t, metricsListener.Addr().String()) + if err = apiListener.Close(); err != nil { + t.Fatal(err) + } + + select { + case err = <-result: + if err == nil || !strings.Contains(err.Error(), "API server stopped unexpectedly") { + t.Fatalf("error = %v, want unexpected API server failure", err) + } + case <-time.After(2 * time.Second): + t.Fatal("server coordinator did not return") + } + + select { + case <-signalsRestored: + default: + t.Error("signal handling was not restored") + } + select { + case <-tasksWaited: + default: + t.Error("background tasks were not awaited") + } + + connection, dialErr := net.DialTimeout("tcp", metricsListener.Addr().String(), 100*time.Millisecond) + if dialErr == nil { + _ = connection.Close() + t.Error("metrics sibling still accepts connections") + } +} + +func TestServeHTTPServersShutsDownOnSignal(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + + server := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusNoContent) + })} + t.Cleanup(func() { + _ = server.Close() + _ = listener.Close() + }) + + sigchan := make(chan os.Signal, 1) + signalsRestored := make(chan struct{}) + tasksWaitedAfterShutdown := make(chan bool, 1) + result := make(chan error, 1) + go func() { + result <- serveHTTPServers( + []apiHTTPServer{{name: "API", server: server, listener: listener}}, + sigchan, + func() { close(signalsRestored) }, + func() { + connection, dialErr := net.DialTimeout("tcp", listener.Addr().String(), 100*time.Millisecond) + if dialErr == nil { + _ = connection.Close() + } + tasksWaitedAfterShutdown <- dialErr != nil + }, + time.Second, + ) + }() + + waitForHTTP(t, listener.Addr().String()) + sigchan <- syscall.SIGTERM + + select { + case err = <-result: + if err != nil { + t.Fatalf("signal shutdown returned error: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("server coordinator did not return") + } + + select { + case <-signalsRestored: + default: + t.Error("signal handling was not restored") + } + select { + case listenerClosed := <-tasksWaitedAfterShutdown: + if !listenerClosed { + t.Error("background tasks were awaited before HTTP shutdown") + } + case <-time.After(time.Second): + t.Error("background tasks were not awaited") + } +} + +func waitForHTTP(t *testing.T, address string) { + t.Helper() + client := &http.Client{Timeout: 100 * time.Millisecond} + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + response, err := client.Get("http://" + address) + if err == nil { + _, _ = io.Copy(io.Discard, response.Body) + _ = response.Body.Close() + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("HTTP server at %s did not become ready", address) +} diff --git a/completion.d/_aptly b/completion.d/_aptly index 6a70dfe3c..f527f129b 100644 --- a/completion.d/_aptly +++ b/completion.d/_aptly @@ -537,6 +537,7 @@ local keyring="*-keyring=[gpg keyring to use when verifying Release file (could serve) _arguments '1:: :' \ "-listen=[host:port for HTTP listening or unix://path to listen on a Unix domain socket]:host\:port or unix\://path: " \ + "-metrics-listen=[host:port for optional metrics-only HTTP listening]:host\:port: " \ "-no-lock=[don’t lock the database]:$bool" ;; esac diff --git a/completion.d/aptly b/completion.d/aptly index bdfbe722c..f1726f938 100644 --- a/completion.d/aptly +++ b/completion.d/aptly @@ -674,7 +674,7 @@ _aptly() "serve") if [[ $numargs -eq 0 ]]; then if [[ "$cur" == -* ]]; then - COMPREPLY=($(compgen -W "-listen=" -- ${cur})) + COMPREPLY=($(compgen -W "-listen= -metrics-listen=" -- ${cur})) fi return 0 fi diff --git a/man/aptly.1 b/man/aptly.1 index 67351d7a7..ffb870b86 100644 --- a/man/aptly.1 +++ b/man/aptly.1 @@ -2437,10 +2437,13 @@ host:port for HTTP listening Start HTTP server with aptly REST API\. The server can listen to either a port or Unix domain socket\. When using a socket, Aptly will fully manage the socket file\. This command also supports taking over from a systemd file descriptors to enable systemd socket activation\. . .P +An optional metrics\-only TCP listener can be enabled with \-metrics\-listen\. It serves GET and HEAD requests at /metrics and requires enableMetricsEndpoint in the aptly configuration\. This listener uses plain HTTP; provide access control and TLS through network policy or a proxy when required\. +. +.P Example: . .P -$ aptly api serve \-listen=:8080 $ aptly api serve \-listen=unix:///tmp/aptly\.sock +$ aptly api serve \-listen=:8080 $ aptly api serve \-listen=unix:///tmp/aptly\.sock $ aptly api serve \-listen=:8080 \-metrics\-listen=127\.0\.0\.1:9090 . .P Options: @@ -2450,6 +2453,10 @@ Options: host:port for HTTP listening or unix://path to listen on a Unix domain socket . .TP +\-\fBmetrics\-listen\fR= +host:port for optional metrics\-only HTTP listening (requires enableMetricsEndpoint) +. +.TP \-\fBno\-lock\fR don\(cqt lock the database . diff --git a/system/t12_api/metrics_listener.py b/system/t12_api/metrics_listener.py new file mode 100644 index 000000000..480832317 --- /dev/null +++ b/system/t12_api/metrics_listener.py @@ -0,0 +1,155 @@ +import http.client +import os +import socket +import subprocess +import time + +from lib import BaseTest +from testout import TestOut + +try: + import requests +except ImportError: + requests = None + + +class MetricsListenerAPITest(BaseTest): + """ + Run the API and metrics-only listeners in one Aptly process. + """ + + aptly_server = None + aptly_out = None + debugOutput = True + + def fixture_available(self): + return super().fixture_available() and requests is not None + + def prepare(self): + super().prepare() + + config_path = os.path.join(os.environ["HOME"], self.aptlyConfigFile) + last_output = "" + + for _ in range(3): + with socket.socket() as api_socket, socket.socket() as metrics_socket: + api_socket.bind(("127.0.0.1", 0)) + metrics_socket.bind(("127.0.0.1", 0)) + self.api_host, self.api_port = api_socket.getsockname() + self.metrics_host, self.metrics_port = metrics_socket.getsockname() + + self.api_url = f"{self.api_host}:{self.api_port}" + self.metrics_url = f"{self.metrics_host}:{self.metrics_port}" + self.aptly_out = TestOut() + self.aptly_server = self._start_process( + f"aptly api serve -no-lock -config={config_path} " + f"-listen={self.api_url} -metrics-listen={self.metrics_url}", + stdout=self.aptly_out, + stderr=self.aptly_out, + ) + + try: + if self._wait_until_ready(): + return + last_output = self.aptly_out.get_contents() + except Exception as error: + self._stop_server() + last_output = self.aptly_out.get_contents() + self._close_output() + raise RuntimeError(f"{error}:\n{last_output}") from error + + self._stop_server() + self._close_output() + + raise RuntimeError(f"Aptly exited before its API listener became ready:\n{last_output}") + + def run(self): + pass + + def teardown(self): + self._stop_server() + self._close_output() + super().teardown() + + def debug_output(self): + if self.aptly_out is None: + return "" + return self.aptly_out.get_contents() + + def _stop_server(self): + if self.aptly_server is None: + return + self.aptly_server.terminate() + try: + self.aptly_server.wait(timeout=10) + except subprocess.TimeoutExpired: + self.aptly_server.kill() + self.aptly_server.wait() + self.aptly_server = None + + def _close_output(self): + if self.aptly_out is not None: + self.aptly_out.close() + self.aptly_out = None + + def _wait_until_ready(self): + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + if self.aptly_server.poll() is not None: + return False + try: + response = requests.get(f"http://{self.api_url}/api/version", timeout=0.2) + if response.status_code == 200: + return True + except requests.RequestException: + pass + time.sleep(0.05) + raise RuntimeError("Aptly API listener did not become ready") + + def check(self): + response = requests.get(f"http://{self.api_url}/api/metrics", timeout=5) + self.check_equal(response.status_code, 200) + self.check_in("# TYPE aptly_build_info gauge", response.text) + + response = requests.head(f"http://{self.api_url}/api/metrics", timeout=5) + self.check_equal(response.status_code, 200) + self.check_equal(response.content, b"") + + response = requests.get( + f"http://{self.metrics_url}/metrics?_async=true&async=true", + timeout=5, + ) + self.check_equal(response.status_code, 200) + self.check_in("# TYPE aptly_build_info gauge", response.text) + self.check_in("# TYPE aptly_api_http_requests_total counter", response.text) + + response = requests.head(f"http://{self.metrics_url}/metrics", timeout=5) + self.check_equal(response.status_code, 200) + self.check_equal(response.content, b"") + + for path in ("/api/metrics", "/api/version", "/metrics/", "/METRICS"): + response = requests.get(f"http://{self.metrics_url}{path}", timeout=5) + self.check_equal(response.status_code, 404) + + connection = http.client.HTTPConnection(self.metrics_host, self.metrics_port, timeout=5) + connection.request("GET", "/met%72ics") + response = connection.getresponse() + self.check_equal(response.status, 404) + response.read() + connection.close() + + response = requests.post(f"http://{self.metrics_url}/metrics", timeout=5) + self.check_equal(response.status_code, 405) + self.check_equal(response.headers.get("Allow"), "GET, HEAD") + + self.aptly_server.terminate() + return_code = self.aptly_server.wait(timeout=10) + self.check_equal(return_code, 0) + self.aptly_server = None + + for url in (self.api_url, self.metrics_url): + try: + requests.get(f"http://{url}/metrics", timeout=0.2) + except requests.RequestException: + continue + raise AssertionError(f"listener {url} still accepts requests after shutdown") From d36c3c2d02877801fbfb3ee02dadcfde6975a318 Mon Sep 17 00:00:00 2001 From: Luan Dang <167596399+luanmdang@users.noreply.github.com> Date: Thu, 10 Sep 2026 13:02:02 -0500 Subject: [PATCH 2/3] Refactor TestHeadMetrics for improved testing Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- api/api_test.go | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/api/api_test.go b/api/api_test.go index 986d131c9..9f02fde3a 100644 --- a/api/api_test.go +++ b/api/api_test.go @@ -159,10 +159,20 @@ func (s *APISuite) TestGetMetrics(c *C) { } func (s *APISuite) TestHeadMetrics(c *C) { - response, err := s.HTTPRequest("HEAD", "/api/metrics", nil) + server := httptest.NewServer(s.router) + defer server.Close() + + req, err := http.NewRequest(http.MethodHead, server.URL+"/api/metrics", nil) c.Assert(err, IsNil) - c.Check(response.Code, Equals, 200) - c.Check(strings.Contains(response.Body.String(), "# TYPE aptly_build_info gauge"), Equals, true) + + resp, err := server.Client().Do(req) + c.Assert(err, IsNil) + defer resp.Body.Close() + + c.Check(resp.StatusCode, Equals, 200) + body, readErr := io.ReadAll(resp.Body) + c.Assert(readErr, IsNil) + c.Check(len(body), Equals, 0) } func (s *APISuite) TestRepoCreate(c *C) { From e8d161f98508ee0ce201e1cba506a2e198cf09f6 Mon Sep 17 00:00:00 2001 From: Luan Dang Date: Thu, 10 Sep 2026 13:11:45 -0500 Subject: [PATCH 3/3] test: handle HEAD metrics response cleanup --- api/api_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api/api_test.go b/api/api_test.go index 9f02fde3a..2b44e9945 100644 --- a/api/api_test.go +++ b/api/api_test.go @@ -167,7 +167,7 @@ func (s *APISuite) TestHeadMetrics(c *C) { resp, err := server.Client().Do(req) c.Assert(err, IsNil) - defer resp.Body.Close() + defer func() { _ = resp.Body.Close() }() c.Check(resp.StatusCode, Equals, 200) body, readErr := io.ReadAll(resp.Body)