atc: behaviour: emit tasks waiting prometheus metric

.github/auto_pr_team.yml

@@ -0,0 +1,2 @@
+org: concourse
+team: contributors

CONTRIBUTING.md

@@ -565,6 +565,43 @@ Kubernetes-related testing are all end-to-end, living under `topgun/k8s`. They
 require access to a real Kubernetes cluster with access granted through a
 properly configured `~/.kube/config` file.
 
+[`kind`] is a great choice when it comes to running a local Kubernetes cluster -
+all you need is `docker`, and the `kind` CLI. If you wish to run the tests with
+a high degree of concurrency, it's advised to have multiple kubernetes nodes.
+This can be achieved with the following `kind` config:
+
+```yaml
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+nodes:
+- role: control-plane
+- role: worker
+- role: worker
+- role: worker
+- role: worker
+```
+
+
+With the cluster up, the next step is to have a proper [Tiller] setup (the tests
+still run with Helm 2):
+
+
+```bash
+kubectl create serviceaccount \
+	--namespace kube-system \
+	tiller
+
+kubectl create clusterrolebinding \
+	tiller-cluster-rule \
+	--clusterrole=cluster-admin \
+	--serviceaccount=kube-system:tiller
+
+helm init \
+	--service-account tiller \
+	--upgrade
+```
+
+
 The tests require a few environment variables to be set:
 
 - `CONCOURSE_IMAGE_TAG` or `CONCOURSE_IMAGE_DIGEST`: the tag or digest to use
@@ -576,12 +613,20 @@ to define the postgres chart that Concourse depends on.
 - `CONCOURSE_CHART_DIR`: location in the filesystem where a copy of [`the Concourse Helm
 chart`][concourse-helm-chart] exists.
 
+
 With those set, go to `topgun/k8s` and run Ginkgo:
 
 ```sh
+# run the test cases serially
 ginkgo .
+
+# run the test cases with a concurrency level of 16
+ginkgo -nodes=16 .
 ```
 
+[`kind`]: https://kind.sigs.k8s.io/
+[Tiller]: https://v2.helm.sh/docs/install/
+
 
 ### A note on `topgun`
 

atc/api/accessor/accessor.go

@@ -231,12 +231,17 @@ func (a *access) connectorID() string {
 }
 
 func (a *access) groups() []string {
+	groups := []string{}
 	if raw, ok := a.claims()["groups"]; ok {
-		if claim, ok := raw.([]string); ok {
-			return claim
+		if rawGroups, ok := raw.([]interface{}); ok {
+			for _, rawGroup := range rawGroups {
+				if group, ok := rawGroup.(string); ok {
+					groups = append(groups, group)
+				}
+			}
 		}
 	}
-	return []string{}
+	return groups
 }
 
 func (a *access) adminTeams() []string {

atc/api/accessor/accessor_test.go

@@ -222,7 +222,7 @@ var _ = Describe("Accessor", func() {
 			verification.HasToken = true
 			verification.IsTokenValid = true
 			verification.RawClaims = map[string]interface{}{
-				"groups": []string{"some-group"},
+				"groups": []interface{}{"some-group"},
 				"federated_claims": map[string]interface{}{
 					"connector_id": "some-connector",
 				},
@@ -649,7 +649,7 @@ var _ = Describe("Accessor", func() {
 						"user_id":      "some-user-id",
 						"user_name":    "some-user-name",
 					},
-					"groups": []string{"some-group"},
+					"groups": []interface{}{"some-group"},
 				}
 			})
 

atc/atccmd/command.go

@@ -93,6 +93,9 @@ var retryingDriverName = "too-many-connections-retrying"
 var flyClientID = "fly"
 var flyClientSecret = "Zmx5"
 
+var workerAvailabilityPollingInterval = 5 * time.Second
+var workerStatusPublishInterval = 1 * time.Minute
+
 type ATCCommand struct {
 	RunCommand RunCommand `command:"run"`
 	Migration  Migration  `command:"migrate"`
@@ -643,7 +646,7 @@ func (cmd *RunCommand) constructAPIMembers(
 	)
 
 	pool := worker.NewPool(workerProvider)
-	workerClient := worker.NewClient(pool, workerProvider, compressionLib)
+	workerClient := worker.NewClient(pool, workerProvider, compressionLib, workerAvailabilityPollingInterval, workerStatusPublishInterval)
 
 	credsManagers := cmd.CredentialManagers
 	dbPipelineFactory := db.NewPipelineFactory(dbConn, lockFactory)
@@ -871,7 +874,11 @@ func (cmd *RunCommand) constructBackendMembers(
 	)
 
 	pool := worker.NewPool(workerProvider)
-	workerClient := worker.NewClient(pool, workerProvider, compressionLib)
+	workerClient := worker.NewClient(pool,
+		workerProvider,
+		compressionLib,
+		workerAvailabilityPollingInterval,
+		workerStatusPublishInterval)
 
 	defaultLimits, err := cmd.parseDefaultLimits()
 	if err != nil {

atc/db/check_factory.go

@@ -3,7 +3,6 @@ package db
 import (
 	"database/sql"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"time"
 
@@ -143,7 +142,7 @@ func (c *checkFactory) TryCreateCheck(logger lager.Logger, checkable Checkable,
 	parentType, found := resourceTypes.Parent(checkable)
 	if found {
 		if parentType.Version() == nil {
-			return nil, false, errors.New("parent type has no version")
+			return nil, false, fmt.Errorf("resource type '%s' has no version", parentType.Name())
 		}
 	}
 

atc/db/log_conn.go

@@ -1,13 +1,16 @@
 package db
 
 import (
+	"context"
 	"database/sql"
 	"strings"
 
 	"code.cloudfoundry.org/lager"
 	"github.com/Masterminds/squirrel"
 )
 
+// Log returns a wrapper of DB connection which contains a wraper of DB transactions
+// so all queries could be logged by givin logger
 func Log(logger lager.Logger, conn Conn) Conn {
 	return &logConn{
 		Conn:   conn,
@@ -21,21 +24,56 @@ type logConn struct {
 	logger lager.Logger
 }
 
+func (c *logConn) Begin() (Tx, error) {
+	tx, err := c.Conn.Begin()
+	if err != nil {
+		return nil, err
+	}
+
+	return &logDbTx{Tx: tx, logger: c.logger}, nil
+}
+
 func (c *logConn) Query(query string, args ...interface{}) (*sql.Rows, error) {
-	c.logger.Debug("query", lager.Data{"query": c.strip(query)})
+	c.logger.Debug("query", lager.Data{"query": strip(query)})
 	return c.Conn.Query(query, args...)
 }
 
 func (c *logConn) QueryRow(query string, args ...interface{}) squirrel.RowScanner {
-	c.logger.Debug("query-row", lager.Data{"query": c.strip(query)})
+	c.logger.Debug("query-row", lager.Data{"query": strip(query)})
 	return c.Conn.QueryRow(query, args...)
 }
 
 func (c *logConn) Exec(query string, args ...interface{}) (sql.Result, error) {
-	c.logger.Debug("exec", lager.Data{"query": c.strip(query)})
+	c.logger.Debug("exec", lager.Data{"query": strip(query)})
 	return c.Conn.Exec(query, args...)
 }
 
-func (c *logConn) strip(query string) string {
+func strip(query string) string {
 	return strings.Join(strings.Fields(query), " ")
 }
+
+type logDbTx struct {
+	Tx
+
+	logger lager.Logger
+}
+
+func (t *logDbTx) Query(query string, args ...interface{}) (*sql.Rows, error) {
+	t.logger.Debug("tx-query", lager.Data{"query": strip(query)})
+	return t.Tx.Query(query, args...)
+}
+
+func (t *logDbTx) QueryRow(query string, args ...interface{}) squirrel.RowScanner {
+	t.logger.Debug("tx-query-row", lager.Data{"query": strip(query)})
+	return t.Tx.QueryRow(query, args...)
+}
+
+func (t *logDbTx) Exec(query string, args ...interface{}) (sql.Result, error) {
+	t.logger.Debug("tx-exec", lager.Data{"query": strip(query)})
+	return t.Tx.Exec(query, args...)
+}
+
+func (t *logDbTx) QueryRowContext(ctx context.Context, query string, args ...interface{}) squirrel.RowScanner {
+	t.logger.Debug("tx-query-row-context", lager.Data{"query": strip(query)})
+	return t.Tx.QueryRowContext(ctx, query, args...)
+}

atc/metric/emitter/newrelic.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"compress/gzip"
 	"encoding/json"
+	"fmt"
 	"io"
 	"io/ioutil"
 	"net/http"
@@ -39,6 +40,7 @@ type (
 	NewRelicConfig struct {
 		AccountID          string        `long:"newrelic-account-id" description:"New Relic Account ID"`
 		APIKey             string        `long:"newrelic-api-key" description:"New Relic Insights API Key"`
+		Url                string        `long:"newrelic-insights-api-url" default:"https://insights-collector.newrelic.com" description:"Base Url for insights Insert API"`
 		ServicePrefix      string        `long:"newrelic-service-prefix" default:"" description:"An optional prefix for emitted New Relic events"`
 		BatchSize          uint64        `long:"newrelic-batch-size" default:"2000" description:"Number of events to batch together before emitting"`
 		BatchDuration      time.Duration `long:"newrelic-batch-duration" default:"60s" description:"Length of time to wait between emitting until all currently batched events are emitted"`
@@ -65,7 +67,7 @@ func (config *NewRelicConfig) NewEmitter() (metric.Emitter, error) {
 
 	return &NewRelicEmitter{
 		Client:             client,
-		Url:                "https://insights-collector.newrelic.com/v1/accounts/" + config.AccountID + "/events",
+		Url:                fmt.Sprintf("%s/v1/accounts/%s/events", config.Url, config.AccountID),
 		apikey:             config.APIKey,
 		prefix:             config.ServicePrefix,
 		containers:         new(stats),

atc/metric/emitter/newrelic_test.go

@@ -3,6 +3,7 @@ package emitter_test
 import (
 	"compress/gzip"
 	"encoding/json"
+	"fmt"
 	"io/ioutil"
 	"net/http"
 	"time"
@@ -142,6 +143,25 @@ var _ = Describe("NewRelicEmitter", func() {
 			Entry("is disabled", true, ""),
 		)
 	})
+
+	Context("NewRelicConfiguration", func() {
+		It("sends events to configured endpoint", func() {
+			config := &emitter.NewRelicConfig{
+				AccountID: "123456",
+				APIKey:    "eu019347923874648573934074",
+				Url:       server.URL(),
+			}
+
+			server.RouteToHandler(http.MethodPost, "/v1/accounts/123456/events", verifyEvents(1))
+
+			e, _ := config.NewEmitter()
+			e.Emit(testLogger, testEvent)
+
+			newRelicEmitter := e.(*emitter.NewRelicEmitter)
+			Expect(newRelicEmitter.Url).To(Equal(fmt.Sprintf("%s/v1/accounts/123456/events", server.URL())))
+			Eventually(server.ReceivedRequests).Should(HaveLen(1))
+		})
+	})
 })
 
 func verifyEvents(expectedEvents int) http.HandlerFunc {

atc/metric/emitter/prometheus.go

@@ -27,6 +27,8 @@ type PrometheusEmitter struct {
 	concurrentRequestsLimitHit *prometheus.CounterVec
 	concurrentRequests         *prometheus.GaugeVec
 
+	tasksWaiting prometheus.Gauge
+
 	buildDurationsVec *prometheus.HistogramVec
 	buildsAborted     prometheus.Counter
 	buildsErrored     prometheus.Counter
@@ -169,6 +171,14 @@ func (config *PrometheusConfig) NewEmitter() (metric.Emitter, error) {
 	}, []string{"action"})
 	prometheus.MustRegister(concurrentRequests)
 
+	tasksWaiting := prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: "concourse",
+		Subsystem: "tasks",
+		Name:      "waiting",
+		Help:      "Number of Concourse tasks currently waiting.",
+	})
+	prometheus.MustRegister(tasksWaiting)
+
 	buildsFinished := prometheus.NewCounter(prometheus.CounterOpts{
 		Namespace: "concourse",
 		Subsystem: "builds",
@@ -388,6 +398,8 @@ func (config *PrometheusConfig) NewEmitter() (metric.Emitter, error) {
 		concurrentRequestsLimitHit: concurrentRequestsLimitHit,
 		concurrentRequests:         concurrentRequests,
 
+		tasksWaiting: tasksWaiting,
+
 		buildDurationsVec: buildDurationsVec,
 		buildsAborted:     buildsAborted,
 		buildsErrored:     buildsErrored,
@@ -452,6 +464,8 @@ func (emitter *PrometheusEmitter) Emit(logger lager.Logger, event metric.Event)
 		emitter.concurrentRequestsLimitHit.WithLabelValues(event.Attributes["action"]).Add(event.Value)
 	case "concurrent requests":
 		emitter.concurrentRequests.WithLabelValues(event.Attributes["action"]).Set(event.Value)
+	case "tasks waiting":
+		emitter.tasksWaiting.Set(event.Value)
 	case "build finished":
 		emitter.buildFinishedMetrics(logger, event)
 	case "worker containers":