atc/api: limit concurrent requests

atc/atccmd/command.go

@@ -119,8 +119,9 @@ type RunCommand struct {
 
 	Postgres flag.PostgresConfig `group:"PostgreSQL Configuration" namespace:"postgres"`
 
-	APIMaxOpenConnections     int `long:"api-max-conns" description:"The maximum number of open connections for the api connection pool." default:"10"`
-	BackendMaxOpenConnections int `long:"backend-max-conns" description:"The maximum number of open connections for the backend connection pool." default:"50"`
+	ConcurrentRequestLimits   map[wrappa.LimitedRoute]int `long:"concurrent-request-limit" description:"Limit the number of concurrent requests to an API endpoint (Example: ListAllJobs:5)"`
+	APIMaxOpenConnections     int                         `long:"api-max-conns" description:"The maximum number of open connections for the api connection pool." default:"10"`
+	BackendMaxOpenConnections int                         `long:"backend-max-conns" description:"The maximum number of open connections for the backend connection pool." default:"50"`
 
 	CredentialManagement creds.CredentialManagementConfig `group:"Credential Management"`
 	CredentialManagers   creds.Managers
@@ -1790,6 +1791,10 @@ func (cmd *RunCommand) constructAPIHandler(
 	}
 
 	apiWrapper := wrappa.MultiWrappa{
+		wrappa.NewConcurrentRequestLimitsWrappa(
+			logger,
+			wrappa.NewConcurrentRequestPolicy(cmd.ConcurrentRequestLimits),
+		),
 		wrappa.NewAPIMetricsWrappa(logger),
 		wrappa.NewAPIAuthWrappa(
 			checkPipelineAccessHandlerFactory,

atc/atccmd/command_test.go

@@ -1,8 +1,10 @@
 package atccmd_test
 
 import (
+	"fmt"
 	"testing"
 
+	"github.com/concourse/concourse/atc"
 	"github.com/concourse/concourse/atc/atccmd"
 	"github.com/jessevdk/go-flags"
 	"github.com/stretchr/testify/require"
@@ -27,6 +29,22 @@ func (s *CommandSuite) TestLetsEncryptDefaultIsUpToDate() {
 	s.Equal(opt.Default, []string{autocert.DefaultACMEDirectory})
 }
 
+func (s *CommandSuite) TestInvalidConcurrentRequestLimitAction() {
+	cmd := &atccmd.RunCommand{}
+	parser := flags.NewParser(cmd, flags.None)
+	_, err := parser.ParseArgs([]string{
+		"--client-secret",
+		"client-secret",
+		"--concurrent-request-limit",
+		fmt.Sprintf("%s:2", atc.GetInfo),
+	})
+
+	s.Contains(
+		err.Error(),
+		fmt.Sprintf("action '%s' is not supported", atc.GetInfo),
+	)
+}
+
 func TestSuite(t *testing.T) {
 	suite.Run(t, &CommandSuite{
 		Assertions: require.New(t),

atc/integration/concurrent_request_limits_test.go

@@ -0,0 +1,32 @@
+package integration_test
+
+import (
+	"net/http"
+
+	"github.com/concourse/concourse/atc"
+	"github.com/concourse/concourse/atc/wrappa"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Concurrent request limits", func() {
+	BeforeEach(func() {
+		cmd.ConcurrentRequestLimits = map[wrappa.LimitedRoute]int{
+			wrappa.LimitedRoute(atc.ListAllJobs): 0,
+		}
+	})
+
+	It("disables ListAllJobs requests", func() {
+		client := login(atcURL, "test", "test")
+		httpClient := client.HTTPClient()
+		request, _ := http.NewRequest(
+			"GET",
+			client.URL()+"/api/v1/jobs",
+			nil,
+		)
+
+		response, _ := httpClient.Do(request)
+
+		Expect(response.StatusCode).To(Equal(http.StatusNotImplemented))
+	})
+})

atc/metric/emitter/newrelic.go

@@ -92,6 +92,8 @@ func (emitter *NewRelicEmitter) Emit(logger lager.Logger, event metric.Event) {
 		"checks queue size",
 		"worker containers",
 		"worker volumes",
+		"concurrent requests",
+		"concurrent requests limit hit",
 		"http response time",
 		"database queries",
 		"database connections",

atc/metric/emitter/prometheus.go

@@ -24,6 +24,9 @@ type PrometheusEmitter struct {
 	buildsStarted prometheus.Counter
 	buildsRunning prometheus.Gauge
 
+	concurrentRequestsLimitHit *prometheus.CounterVec
+	concurrentRequests         *prometheus.GaugeVec
+
 	buildDurationsVec *prometheus.HistogramVec
 	buildsAborted     prometheus.Counter
 	buildsErrored     prometheus.Counter
@@ -151,6 +154,21 @@ func (config *PrometheusConfig) NewEmitter() (metric.Emitter, error) {
 	})
 	prometheus.MustRegister(buildsRunning)
 
+	concurrentRequestsLimitHit := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "concourse",
+		Subsystem: "concurrent_requests",
+		Name:      "limit_hit_total",
+		Help:      "Total number of requests rejected because the server was already serving too many concurrent requests.",
+	}, []string{"action"})
+	prometheus.MustRegister(concurrentRequestsLimitHit)
+
+	concurrentRequests := prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: "concourse",
+		Name:      "concurrent_requests",
+		Help:      "Number of concurrent requests being served by endpoints that have a specified limit of concurrent requests.",
+	}, []string{"action"})
+	prometheus.MustRegister(concurrentRequests)
+
 	buildsFinished := prometheus.NewCounter(prometheus.CounterOpts{
 		Namespace: "concourse",
 		Subsystem: "builds",
@@ -367,6 +385,9 @@ func (config *PrometheusConfig) NewEmitter() (metric.Emitter, error) {
 		buildsStarted: buildsStarted,
 		buildsRunning: buildsRunning,
 
+		concurrentRequestsLimitHit: concurrentRequestsLimitHit,
+		concurrentRequests:         concurrentRequests,
+
 		buildDurationsVec: buildDurationsVec,
 		buildsAborted:     buildsAborted,
 		buildsErrored:     buildsErrored,
@@ -427,6 +448,10 @@ func (emitter *PrometheusEmitter) Emit(logger lager.Logger, event metric.Event)
 		emitter.buildsStarted.Add(event.Value)
 	case "builds running":
 		emitter.buildsRunning.Set(event.Value)
+	case "concurrent requests limit hit":
+		emitter.concurrentRequestsLimitHit.WithLabelValues(event.Attributes["action"]).Add(event.Value)
+	case "concurrent requests":
+		emitter.concurrentRequests.WithLabelValues(event.Attributes["action"]).Set(event.Value)
 	case "build finished":
 		emitter.buildFinishedMetrics(logger, event)
 	case "worker containers":

atc/metric/metrics.go

@@ -36,6 +36,9 @@ var ChecksQueueSize = &Gauge{}
 var ChecksStarted = &Counter{}
 var ChecksEnqueued = &Counter{}
 
+var ConcurrentRequests = map[string]*Gauge{}
+var ConcurrentRequestsLimitHit = map[string]*Counter{}
+
 type BuildCollectorDuration struct {
 	Duration time.Duration
 }

atc/metric/periodic.go

@@ -139,6 +139,32 @@ func tick(logger lager.Logger) {
 		},
 	)
 
+	for action, gauge := range ConcurrentRequests {
+		emit(
+			logger.Session("concurrent-requests"),
+			Event{
+				Name:  "concurrent requests",
+				Value: gauge.Max(),
+				Attributes: map[string]string{
+					"action": action,
+				},
+			},
+		)
+	}
+
+	for action, counter := range ConcurrentRequestsLimitHit {
+		emit(
+			logger.Session("concurrent-requests-limit-hit"),
+			Event{
+				Name:  "concurrent requests limit hit",
+				Value: counter.Delta(),
+				Attributes: map[string]string{
+					"action": action,
+				},
+			},
+		)
+	}
+
 	emit(
 		logger.Session("checks-finished-with-error"),
 		Event{

atc/metric/periodic_test.go

@@ -30,14 +30,17 @@ var _ = Describe("Periodic emission of metrics", func() {
 		metric.RegisterEmitter(emitterFactory)
 		emitterFactory.IsConfiguredReturns(true)
 		emitterFactory.NewEmitterReturns(emitter, nil)
-		a := &dbfakes.FakeConn{}
-		a.NameReturns("A")
-		b := &dbfakes.FakeConn{}
-		b.NameReturns("B")
-		metric.Databases = []db.Conn{a, b}
 		metric.Initialize(testLogger, "test", map[string]string{}, 1000)
 
-		process = ifrit.Invoke(metric.PeriodicallyEmit(lager.NewLogger("dont care"), 250*time.Millisecond))
+	})
+
+	JustBeforeEach(func() {
+		runner := metric.PeriodicallyEmit(
+			lager.NewLogger("dont care"),
+			250*time.Millisecond,
+		)
+
+		process = ifrit.Invoke(runner)
 	})
 
 	AfterEach(func() {
@@ -46,38 +49,95 @@ var _ = Describe("Periodic emission of metrics", func() {
 		metric.Deinitialize(nil)
 	})
 
-	It("emits database queries", func() {
-		Eventually(emitter.EmitCallCount).Should(BeNumerically(">=", 1))
-		Expect(emitter.Invocations()["Emit"]).To(
-			ContainElement(
+	Context("database-related metrics", func() {
+		BeforeEach(func() {
+			a := &dbfakes.FakeConn{}
+			a.NameReturns("A")
+			b := &dbfakes.FakeConn{}
+			b.NameReturns("B")
+			metric.Databases = []db.Conn{a, b}
+		})
+
+		It("emits database queries", func() {
+			Eventually(emitter.EmitCallCount).Should(BeNumerically(">=", 1))
+			Expect(emitter.Invocations()["Emit"]).To(
 				ContainElement(
-					MatchFields(IgnoreExtras, Fields{
-						"Name": Equal("database queries"),
-					}),
+					ContainElement(
+						MatchFields(IgnoreExtras, Fields{
+							"Name": Equal("database queries"),
+						}),
+					),
 				),
-			),
-		)
+			)
 
-		By("emits database connections for each pool")
-		Expect(emitter.Invocations()["Emit"]).To(
-			ContainElement(
+			By("emits database connections for each pool")
+			Expect(emitter.Invocations()["Emit"]).To(
 				ContainElement(
-					MatchFields(IgnoreExtras, Fields{
-						"Name":       Equal("database connections"),
-						"Attributes": Equal(map[string]string{"ConnectionName": "A"}),
-					}),
+					ContainElement(
+						MatchFields(IgnoreExtras, Fields{
+							"Name":       Equal("database connections"),
+							"Attributes": Equal(map[string]string{"ConnectionName": "A"}),
+						}),
+					),
 				),
-			),
-		)
-		Expect(emitter.Invocations()["Emit"]).To(
-			ContainElement(
+			)
+			Expect(emitter.Invocations()["Emit"]).To(
 				ContainElement(
-					MatchFields(IgnoreExtras, Fields{
-						"Name":       Equal("database connections"),
-						"Attributes": Equal(map[string]string{"ConnectionName": "B"}),
-					}),
+					ContainElement(
+						MatchFields(IgnoreExtras, Fields{
+							"Name":       Equal("database connections"),
+							"Attributes": Equal(map[string]string{"ConnectionName": "B"}),
+						}),
+					),
 				),
-			),
-		)
+			)
+		})
+	})
+
+	Context("concurrent requests", func() {
+		const action = "ListAllSomething"
+
+		BeforeEach(func() {
+			gauge := &metric.Gauge{}
+			gauge.Set(123)
+
+			counter := &metric.Counter{}
+			counter.IncDelta(10)
+
+			metric.ConcurrentRequests[action] = gauge
+			metric.ConcurrentRequestsLimitHit[action] = counter
+		})
+
+		It("emits", func() {
+			Eventually(emitter.EmitCallCount).Should(BeNumerically(">=", 1))
+
+			Expect(emitter.Invocations()["Emit"]).To(
+				ContainElement(
+					ContainElement(
+						MatchFields(IgnoreExtras, Fields{
+							"Name":  Equal("concurrent requests"),
+							"Value": Equal(float64(123)),
+							"Attributes": Equal(map[string]string{
+								"action": action,
+							}),
+						}),
+					),
+				),
+			)
+
+			Expect(emitter.Invocations()["Emit"]).To(
+				ContainElement(
+					ContainElement(
+						MatchFields(IgnoreExtras, Fields{
+							"Name":  Equal("concurrent requests limit hit"),
+							"Value": Equal(float64(10)),
+							"Attributes": Equal(map[string]string{
+								"action": action,
+							}),
+						}),
+					),
+				),
+			)
+		})
 	})
 })

atc/routes.go

@@ -82,8 +82,8 @@ const (
 	GetLogLevel = "GetLogLevel"
 
 	DownloadCLI  = "DownloadCLI"
-	GetInfo      = "Info"
-	GetInfoCreds = "InfoCreds"
+	GetInfo      = "GetInfo"
+	GetInfoCreds = "GetInfoCreds"
 
 	ListContainers           = "ListContainers"
 	GetContainer             = "GetContainer"