atc: behaviour: emit tasks waiting prometheus metric

[tenjaa]

Existing Issue

Fixes #5057

Why do we need this PR?

Using the limit-active-tasks placement strategy gives the perfect opportunity
to dynamically scale workers based on tasks running. This metric makes
it easier for operators to implement such a scaling.

Changes proposed in this pull request

• add prometheus metric for waiting builds when using limit-active-tasks placement strategy

Contributor Checklist

• Unit tests
  Not sure if we need tests for Metrics. Looking through the code we didn't find any tests for metrics. Should we add some?
• Updated release notes (located at https://github.com/concourse/concourse/tree/master/release-notes)

Reviewer Checklist

• Code reviewed
• Tests reviewed
• Documentation reviewed
• Release notes reviewed
• PR acceptance performed
• New config flags added? Ensure that they are added to the BOSH and Helm packaging; otherwise, ignored for the integration tests (for example, if they are Garden configs that are not displayed in the --help text).

[jamieklassen]

Initially I had some concerns about the ability to use a gauge (the number of tasks currently waiting, which can increase and decrease) instead of a counter (the number of attempts to schedule a task container which have failed due to all workers being at capacity), because I thought there could be some misbehaviour when multiple web nodes were in play. So I tested it out manually.

diff --git a/docker-compose.yml b/docker-compose.yml
index 5618630a7..aeac1ed14 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -11,7 +11,7 @@ services:
       POSTGRES_USER: dev
       POSTGRES_PASSWORD: dev
 
-  web:
+  web-1:
     build: .
     image: concourse/concourse:local
     volumes:
@@ -20,6 +20,7 @@ services:
     depends_on: [db]
     ports:
     - 8080:8080
+    - 9100:9100
     environment:
       CONCOURSE_LOG_LEVEL: debug
       CONCOURSE_POSTGRES_HOST: db
@@ -32,20 +33,53 @@ services:
       CONCOURSE_CLUSTER_NAME: dev
       CONCOURSE_CLIENT_SECRET: Y29uY291cnNlLXdlYgo=
       CONCOURSE_TSA_CLIENT_SECRET: Y29uY291cnNlLXdvcmtlcgo=
+      CONCOURSE_CONTAINER_PLACEMENT_STRATEGY: limit-active-tasks
+      CONCOURSE_MAX_ACTIVE_TASKS_PER_WORKER: 1
+      CONCOURSE_PROMETHEUS_BIND_IP: "0.0.0.0"
+      CONCOURSE_PROMETHEUS_BIND_PORT: "9100"
+      CONCOURSE_PEER_ADDRESS: web-1
+
+  web-2:
+    build: .
+    image: concourse/concourse:local
+    volumes:
+    - .:/src
+    command: web
+    depends_on: [db]
+    ports:
+    - 8081:8080
+    - 9101:9100
+    environment:
+      CONCOURSE_LOG_LEVEL: debug
+      CONCOURSE_POSTGRES_HOST: db
+      CONCOURSE_POSTGRES_USER: dev
+      CONCOURSE_POSTGRES_PASSWORD: dev
+      CONCOURSE_POSTGRES_DATABASE: concourse
+      CONCOURSE_EXTERNAL_URL: http://localhost:8081
+      CONCOURSE_ADD_LOCAL_USER: test:test,guest:guest
+      CONCOURSE_MAIN_TEAM_LOCAL_USER: test
+      CONCOURSE_CLUSTER_NAME: dev
+      CONCOURSE_CLIENT_SECRET: Y29uY291cnNlLXdlYgo=
+      CONCOURSE_TSA_CLIENT_SECRET: Y29uY291cnNlLXdvcmtlcgo=
+      CONCOURSE_CONTAINER_PLACEMENT_STRATEGY: limit-active-tasks
+      CONCOURSE_MAX_ACTIVE_TASKS_PER_WORKER: 1
+      CONCOURSE_PROMETHEUS_BIND_IP: "0.0.0.0"
+      CONCOURSE_PROMETHEUS_BIND_PORT: "9100"
+      CONCOURSE_PEER_ADDRESS: web-2
 
   worker:
     build: .
     image: concourse/concourse:local
     command: worker
     privileged: true
-    depends_on: [web]
+    depends_on: [web-1,web-2]
     ports:
     - 7777:7777
     - 7788:7788
     stop_signal: SIGUSR2
     environment:
       CONCOURSE_LOG_LEVEL: debug
-      CONCOURSE_TSA_HOST: web:2222
+      CONCOURSE_TSA_HOST: web-1:2222
 
       # avoid using loopbacks
       CONCOURSE_BAGGAGECLAIM_DRIVER: overlay
diff --git a/hack/overrides/prometheus.yml b/hack/overrides/prometheus.yml
index bd4ceb2ae..71f2dc762 100644
--- a/hack/overrides/prometheus.yml
+++ b/hack/overrides/prometheus.yml
@@ -8,13 +8,6 @@
 version: '3'
 
 services:
-  web:
-    environment:
-      CONCOURSE_PROMETHEUS_BIND_IP: "0.0.0.0"
-      CONCOURSE_PROMETHEUS_BIND_PORT: "9100"
-    ports:
-      - '9100:9100'
-
   prometheus:
     image: prom/prometheus
     entrypoint:
@@ -30,7 +23,8 @@ services:
           - job_name: 'concourse'
             static_configs:
               - targets:
-                - 'web:9100'
+                - 'web-1:9100'
+                - 'web-2:9100'
         " > config.yml
 
           exec prometheus \

And what was striking was that it totally worked! I guess the logic that acquires a lock when attempting to schedule a container actually synchronizes through the database -- i was concerned we might "over-report" the number of waiting tasks since two different web nodes might be attempting to start the same task.

I ran

for i in {1..10}; do
  fly -t web2 execute -c web/wats/fixtures/smoke-task.yml -i some-input=web/wats/fixtures/some-input &
  fly -t dev execute -c web/wats/fixtures/smoke-task.yml -i some-input=web/wats/fixtures/some-input &
done

where dev was localhost:8080 and web2 was localhost:8081. And what I saw in prometheus was quite promising!

Each web node schedules its own containers. Now I need to attempt a similar exercise using job builds instead of one-offs (fly execute).

[jamieklassen]

This continues to resist my attempts to break it - similar good behaviour when I trigger a bunch of job builds:

Before I get into it, the implementation itself looks pretty much gold - very idiomatic, small diff, no big WTFs, and it works! Very well done.

I'd love to see some documentation for these metrics, maybe around here (source) or here (source) -- would you mind submitting patches for those?

In terms of testing, I think you could certainly add unit tests for a few things.

• you're periodically forwarding the state of a global gauge to the emitter - there are tests for this behaviour in https://github.com/concourse/concourse/blob/master/atc/metric/periodic_test.go. (full disclosure, I have an open PR that adds tests here, too!)
• you're configuring an entry to be exposed by the prometheus endpoint - there are tests for this sort of thing in https://github.com/concourse/concourse/blob/master/atc/metric/emitter/prometheus_test.go
• something I find very interesting and wasn't obvious to me until I extensively tested manually was the concurrency-safety of this approach. A test around the worker client might have made this super clear! Around https://github.com/concourse/concourse/blob/master/atc/worker/client_test.go#L659-L667 maybe.

Finally, there's a merge conflict in the release notes - probably my bad for taking so long to get to this review. I was crunching on releasing 5.5.11, sorry! Anyway, I'm sure a simple rebase would take care of it.

[tenjaa]

Hi @pivotal-jamie-klassen
we added some tests and also fixed the merge conflict.
We will also create a PR for the docs today (or tomorrow).

[shyamz-22]

@pivotal-jamie-klassen we have added test as mentioned in the last review.

However our test that tests if the task waiting metric is gauged correctly when waiting for worker to be available is bit flaky. It seem to succeed after 2 or 3 attempts. Can you help us with that?
Thanks :)

[jamieklassen]

tl;dr - making this properly testable may require some dependency-breaking refactors. Or maybe a more sophisticated stub on the fakePool's FindOrChooseWorkerForContainer method.

@shyamz-22 I don't have time to go deeply into a solution right now, but I will come back to this and I just want to bring the challenge to the surface.

The full scenario we're trying to simulate is this:

1. the client cannot find a worker
2. the waiting tasks gauge is incremented
3. the step is aborted
4. the waiting tasks gauge is decremented

In the JustBeforeEach, client.RunTaskStep executes, and is blocking. When we look in the implementation, we are getting a race between the context being canceled and the metric being incremented. Somehow we want to wait long enough for the metric to actually be incremented before aborting the step.

For the record, I was able to get the test to consistently pass by writing it like this:

Context("when waiting for worker to be available", func() {
	BeforeEach(func() {
		fakePool.FindOrChooseWorkerForContainerReturns(nil, nil)
		go func() {
			Eventually(metric.TasksWaiting.Max, 6*time.Second).Should(Equal(float64(1)))
			cancel()
		}()
	})

	It("metric task waiting is gauged and lock is released", func() {
		Expect(metric.TasksWaiting.Max()).To(Equal(float64(0)))
		Expect(fakeLock.ReleaseCallCount()).To(Equal(fakeLockFactory.AcquireCallCount()))
	})
})

However, I strongly do not recommend keeping this approach, because instead of failing it just hangs indefinitely. But it serves as an interesting illustration of how to orchestrate the concurrent processes to achieve the simulation described.

[shyamz-22]

@pivotal-jamie-klassen thanks for the reply.

We were also in the same opinion as you. We also had an idea for the refactor to be able to test this behaviour better. Could we explore this refactor? or would you suggest a lead on how we can proceed

thanks

[shyamz-22]

@pivotal-jamie-klassen
Hi,

Thank you so much for the feedback. I have made some changes here.

Please consider this as a draft.

Changes:

1. https://github.com/volkswagen/concourse/blob/0f7d6d1040658596568ba9b6f28fb8a65370041d/atc/worker/client.go#L557
2. https://github.com/volkswagen/concourse/blob/0f7d6d1040658596568ba9b6f28fb8a65370041d/atc/worker/client.go#L112
3. https://github.com/volkswagen/concourse/blob/0f7d6d1040658596568ba9b6f28fb8a65370041d/atc/atccmd/command.go#L648
4. https://github.com/volkswagen/concourse/blob/0f7d6d1040658596568ba9b6f28fb8a65370041d/atc/atccmd/command.go#L876
5. https://github.com/volkswagen/concourse/blob/0f7d6d1040658596568ba9b6f28fb8a65370041d/atc/atccmd/command.go#L149
6. https://github.com/volkswagen/concourse/blob/draft-waiting-tasks-refactor/atc/worker/choose_task_worker_test.go

Some differences to the implementation and the draft here:

Previously the order of execution was

1. Acquire lock
2. Find Existing container
3. Choose worker

However the steps 1 and 2 are needed only when a worker is found. There are no tests to prove that the order matters. The current refactor translates roughly to the following (omitted release lock, you can find it in the code)

1. Choose Worker
2. Acquire lock
3. If worker is found increase active task and exit
4. If worker is not found
   a. increase waiting task metric
   b. wait for worker to be available and continue

Observations:

• Tests are not flaky and faster [~2seconds]
• current gocyclo score is

➜  concourse git:(draft-waiting-tasks-refactor) gocyclo atc/worker/client.go
16 worker (*client).RunTaskStep atc/worker/client.go:249:1
12 worker (*client).chooseTaskWorker atc/worker/client.go:557:1

I also reordered the code in the order struts declaration, vars come first followed by Exported, private Receiver methods and at last functions

Please let me know how to proceed further.

[jamieklassen]

@shyamz-22 I quite like the reorganizing of that file, as well as the various *Dummy() methods in the test. The test is also much easier to read. Just for fun, I checked out the coverage, and while atc/worker/client.go has ~80% coverage, all of the functions involved in RunTaskStep are quite thoroughly covered.

I had one concern, and this mostly shows the limited degree to which I actually understand the affected components of Concourse. By acquiring the lock after choosing the worker, I can imagine this reducing the thread-safety of the worker-choosing routine. Since the worker-choosing part is no longer synchronized, can multiple web nodes race each other and do anything weird?

I think the answer is no, because the only code path that actually calls RunTaskStep is the one involving the Run method on a TaskStep, which ultimately only gets called from engineBuild.Run. This method itself involves acquiring a tracking lock. Since there can only be a single goroutine that possesses the tracking lock for a build, there is no risk of two webs accidentally scheduling the same task on different workers. It seems like the active task lock is really just to avoid two webs racing to increment/decrement the active tasks on a worker, and having that number potentially go out of sync with the actual number of task containers on the worker.

Since you are making the retry and reporting interval configurable, there are some jobs in our pipeline that will complain unless the appropriate configuration is exposed via the helm chart and bosh release. So it would be helpful to make PRs to those repos as well, though I can help with that. Otherwise, the configuration doesn't actually need to be passed all the way to the operator. One could still externalize them from the method under test but have them always be constant in production. I suggest considering this because it seems like the kind of configuration an operator would have no idea how or why to tune -- indeed I can't think of why I would ever change that variable! Maybe you can?

[shyamz-22]

One could still externalize them from the method under test but have them always be constant in production.

My confusion to achieve this are:

1. chooseTaskWorker is not exported and is only called from the RunTaskStep which is an interface method. This method (RunTaskStep) is the method under test which means the signature has to be modified.

2. If we Export the chooseTaskWorker method on the client, The client struct is again not exported and is implementing the Client interface

3. All the test packages are having _test and I do not want to add a test with the same package name as the production code breaking the convention.

This is why I decided to pass it as a configuration. I cannot think of any other way as my understanding of all the components and code convention of Concourse is not strong

[jamieklassen]

@shyamz-22 sorry, I was waving my hands there. What I mean was that instead of adding these parameters you could make a change around here:

https://github.com/volkswagen/concourse/blob/0f7d6d1040658596568ba9b6f28fb8a65370041d/atc/atccmd/command.go#L86-L94

adding some lines defining constants:

var workerAvailabilityPollingInterval = 5 * time.Second
var workerStatusPublishInterval = 1 * time.Minute

and then the code here:

https://github.com/volkswagen/concourse/blob/0f7d6d1040658596568ba9b6f28fb8a65370041d/atc/atccmd/command.go#L648

could instead look like

	workerClient := worker.NewClient(pool, workerProvider, compressionLib, workerAvailabilityPollingInterval, workerStatusPublishInterval)

and similarly here.

Does this make sense? I'm also open to making those things operator-configurable if you think there's a precedent, I just can't think of any.

[shyamz-22]

@pivotal-jamie-klassen I also cannot find a valid reason why an operator would need to configure the polling interval. I will do as you suggested by adding defaults and merge the changes. Hopefully this closes the PR 🤞 😄

[jamieklassen]

Thank you for being so responsive and thorough, and really taking the time for craftsmanship. It makes a huge difference!

[shyamz-22]

@pivotal-jamie-klassen thank you to you too for providing enough information in each review really appreciate it :)