mirror of
https://github.com/aptly-dev/aptly.git
synced 2026-11-13 14:37:18 +00:00
Merge pull request #1637 from luanmdang/api-metrics-listener
api: add metrics-only listener
This commit is contained in:
9 files changed
+849
-14
No files matched your search
@@ -87,3 +87,4 @@ List of contributors, in chronological order:
|
||||
* Kerem Gur (https://github.com/KereMath)
|
||||
* Phil Dibowitz (https://github.com/jaymzh)
|
||||
* James Munson (https://github.com/jmunson)
|
||||
* Luan Dang (https://github.com/luanmdang)
|
||||
@@ -159,6 +159,23 @@ func (s *APISuite) TestGetMetrics(c *C) {
|
||||
c.Check(b, Matches, ".*aptly_build_info.*version=\"testVersion\".*")
|
||||
}
|
||||
|
||||
func (s *APISuite) TestHeadMetrics(c *C) {
|
||||
server := httptest.NewServer(s.router)
|
||||
defer server.Close()
|
||||
|
||||
req, err := http.NewRequest(http.MethodHead, server.URL+"/api/metrics", nil)
|
||||
c.Assert(err, IsNil)
|
||||
|
||||
resp, err := server.Client().Do(req)
|
||||
c.Assert(err, IsNil)
|
||||
defer func() { _ = 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) {
|
||||
body, err := json.Marshal(gin.H{
|
||||
"Name": "dummy",
|
||||
|
||||
+3
-1
@@ -145,7 +145,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)
|
||||
|
||||
+241
-11
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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
|
||||
|
||||
+1
-1
@@ -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
|
||||
|
||||
+8
-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
|
||||
.
|
||||
|
||||
@@ -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")
|
||||
Reference in new issue
Block a user