You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@aurora.apache.org by sa...@apache.org on 2018/06/05 23:16:18 UTC
[1/4] aurora git commit: Enable `Tasks` to specify their own custom
maintenance SLA.
Repository: aurora
Updated Branches:
refs/heads/master 34be63158 -> f2acf53ff
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/python/apache/aurora/client/cli/test_kill.py
----------------------------------------------------------------------
diff --git a/src/test/python/apache/aurora/client/cli/test_kill.py b/src/test/python/apache/aurora/client/cli/test_kill.py
index 0e859dc..bb78ee2 100644
--- a/src/test/python/apache/aurora/client/cli/test_kill.py
+++ b/src/test/python/apache/aurora/client/cli/test_kill.py
@@ -59,7 +59,8 @@ class TestKillCommand(AuroraClientCommandTest):
fake_context = FakeAuroraCommandContext()
fake_context.set_options(mock_options)
-
+ fake_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
mock_api = fake_context.get_api('test')
mock_api.kill_job.return_value = AuroraClientCommandTest.create_blank_response(
ResponseCode.JOB_UPDATING_ERROR, "Error.")
@@ -86,6 +87,8 @@ class TestKillCommand(AuroraClientCommandTest):
fake_context.set_options(mock_options)
fake_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
+ fake_context.add_expected_query_result(
AuroraClientCommandTest.create_query_call_result(
AuroraClientCommandTest.create_scheduled_task(1, ScheduleStatus.RUNNING)))
@@ -115,6 +118,8 @@ class TestKillCommand(AuroraClientCommandTest):
fake_context = FakeAuroraCommandContext()
fake_context.set_options(mock_options)
+ fake_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
fake_context.add_expected_query_result(AuroraClientCommandTest.create_empty_task_result())
with pytest.raises(Context.CommandError) as e:
@@ -131,6 +136,8 @@ class TestKillCommand(AuroraClientCommandTest):
fake_context = FakeAuroraCommandContext()
fake_context.set_options(mock_options)
+ fake_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
fake_context.add_expected_query_result(AuroraClientCommandTest.create_empty_task_result())
command.execute(fake_context)
@@ -147,6 +154,8 @@ class TestKillCommand(AuroraClientCommandTest):
fake_context = FakeAuroraCommandContext()
fake_context.set_options(mock_options)
+ fake_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
fake_context.add_expected_query_result(AuroraClientCommandTest.create_empty_task_result())
command.execute(fake_context)
@@ -170,6 +179,8 @@ class TestKillCommand(AuroraClientCommandTest):
fake_context.add_config(config)
mock_api = fake_context.get_api('test')
+ fake_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
fake_context.add_expected_query_result(AuroraClientCommandTest.create_empty_task_result())
mock_api.kill_job.return_value = self.create_simple_success_response()
@@ -194,6 +205,8 @@ class TestKillCommand(AuroraClientCommandTest):
fake_context.add_config(config)
fake_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
+ fake_context.add_expected_query_result(
AuroraClientCommandTest.create_query_call_result(
AuroraClientCommandTest.create_scheduled_task(1, ScheduleStatus.RUNNING)))
@@ -222,6 +235,8 @@ class TestKillAllCommand(AuroraClientCommandTest):
fake_context.add_config(config)
mock_api = fake_context.get_api('test')
+ fake_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
fake_context.add_expected_query_result(AuroraClientCommandTest.create_empty_task_result())
mock_api.kill_job.return_value = self.create_simple_success_response()
@@ -272,8 +287,9 @@ class TestClientKillCommand(AuroraClientCommandTest):
assert api.kill_job.mock_calls == [call(cls.TEST_JOBKEY, None, config=None, message=None)]
@classmethod
- def assert_query(cls, fake_api):
- calls = [call(TaskQuery(jobKeys=[cls.TEST_JOBKEY.to_thrift()], statuses=ACTIVE_STATES))]
+ def assert_query(cls, fake_api, times=1):
+ calls = [call(TaskQuery(jobKeys=[cls.TEST_JOBKEY.to_thrift()], statuses=ACTIVE_STATES))
+ for _ in range(times)]
assert fake_api.query_no_configs.mock_calls == calls
def test_killall_job(self):
@@ -321,6 +337,8 @@ class TestClientKillCommand(AuroraClientCommandTest):
patch('apache.aurora.client.cli.jobs.Job.create_context', return_value=mock_context),
patch('apache.aurora.client.cli.jobs.JobMonitor', return_value=mock_monitor)) as (_, m):
api = mock_context.get_api('west')
+ mock_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
api.kill_job.return_value = self.create_simple_success_response()
cmd = AuroraCommandLine()
cmd.execute(['job', 'kill', '--no-batching', self.get_instance_spec('0,2,4-6')])
@@ -328,6 +346,7 @@ class TestClientKillCommand(AuroraClientCommandTest):
instances = [0, 2, 4, 5, 6]
self.assert_kill_calls(api, instances=instances, message=None)
self.assert_wait_calls(mock_monitor, m.terminal, instances=instances)
+ self.assert_query(api)
def test_kill_job_with_invalid_instances_strict(self):
"""Test kill client-side API logic."""
@@ -350,12 +369,15 @@ class TestClientKillCommand(AuroraClientCommandTest):
patch('apache.aurora.client.cli.jobs.Job.create_context', return_value=mock_context),
patch('apache.aurora.client.cli.jobs.JobMonitor', return_value=mock_monitor)) as (_, m):
api = mock_context.get_api('west')
+ mock_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
api.kill_job.return_value = self.create_simple_success_response()
cmd = AuroraCommandLine()
cmd.execute(['job', 'kill', '--no-batching', self.get_instance_spec('0,2,4-6,11-13')])
instances = [0, 2, 4, 5, 6, 11, 12, 13]
self.assert_kill_calls(api, instances=instances, message=None)
self.assert_wait_calls(mock_monitor, m.terminal, instances=instances)
+ self.assert_query(api)
def test_kill_job_with_instances_batched(self):
"""Test kill client-side API logic."""
@@ -366,7 +388,9 @@ class TestClientKillCommand(AuroraClientCommandTest):
patch('apache.aurora.client.cli.jobs.JobMonitor', return_value=mock_monitor)) as (_, m):
api = mock_context.get_api('west')
mock_context.add_expected_query_result(
- self.create_query_call_result(), job_key=self.TEST_JOBKEY)
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
+ mock_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
api.kill_job.return_value = self.create_simple_success_response()
cmd = AuroraCommandLine()
@@ -374,7 +398,7 @@ class TestClientKillCommand(AuroraClientCommandTest):
self.assert_kill_calls(api, instance_range=range(7))
self.assert_wait_calls(mock_monitor, m.terminal, instance_range=range(7))
- self.assert_query(api)
+ self.assert_query(api, times=2)
def test_kill_job_with_instances_batched_maxerrors(self):
"""Test kill client-side API logic."""
@@ -385,7 +409,9 @@ class TestClientKillCommand(AuroraClientCommandTest):
patch('apache.aurora.client.cli.jobs.JobMonitor', return_value=mock_monitor)):
api = mock_context.get_api('west')
mock_context.add_expected_query_result(
- self.create_query_call_result(), job_key=self.TEST_JOBKEY)
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
+ mock_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
api.kill_job.return_value = self.create_simple_success_response()
cmd = AuroraCommandLine()
@@ -393,7 +419,7 @@ class TestClientKillCommand(AuroraClientCommandTest):
# We should have aborted after the second batch.
self.assert_kill_calls(api, instance_range=range(2), message=None)
- self.assert_query(api)
+ self.assert_query(api, times=2)
def test_kill_job_with_empty_instances_batched(self):
"""Test kill client-side API logic."""
@@ -401,6 +427,8 @@ class TestClientKillCommand(AuroraClientCommandTest):
with contextlib.nested(
patch('apache.aurora.client.cli.jobs.Job.create_context', return_value=mock_context)):
api = mock_context.get_api('west')
+ mock_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
# set up an empty instance list in the getTasksWithoutConfigs response
status_response = self.create_simple_success_response()
status_response.result = Result(scheduleStatusResult=ScheduleStatusResult(tasks=[]))
@@ -410,6 +438,7 @@ class TestClientKillCommand(AuroraClientCommandTest):
cmd.execute(['job', 'kill', self.get_instance_spec('0,2,4-13')])
assert api.kill_job.call_count == 0
+ self.assert_query(api, times=2)
def test_killall_job_output(self):
"""Test kill output."""
@@ -433,6 +462,8 @@ class TestClientKillCommand(AuroraClientCommandTest):
patch('apache.aurora.client.cli.jobs.Job.create_context', return_value=mock_context),
patch('apache.aurora.client.cli.jobs.JobMonitor', return_value=self.get_monitor_mock())):
api = mock_context.get_api('west')
+ mock_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
mock_context.add_expected_query_result(self.create_query_call_result())
api.kill_job.return_value = self.create_simple_success_response()
cmd = AuroraCommandLine()
@@ -441,6 +472,7 @@ class TestClientKillCommand(AuroraClientCommandTest):
assert mock_context.get_out() == ['Successfully killed instances [0, 2, 4, 5, 6]',
'Job kill succeeded']
assert mock_context.get_err() == []
+ self.assert_query(api, times=2)
def test_kill_job_with_instances_batched_maxerrors_output(self):
"""Test kill client-side API logic."""
@@ -450,6 +482,8 @@ class TestClientKillCommand(AuroraClientCommandTest):
patch('apache.aurora.client.cli.jobs.Job.create_context', return_value=mock_context),
patch('apache.aurora.client.cli.jobs.JobMonitor', return_value=mock_monitor)):
api = mock_context.get_api('west')
+ mock_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
mock_context.add_expected_query_result(self.create_query_call_result())
api.kill_job.return_value = self.create_simple_success_response()
cmd = AuroraCommandLine()
@@ -461,6 +495,7 @@ class TestClientKillCommand(AuroraClientCommandTest):
'Instances [0, 2, 4, 5, 6] were not killed in time',
'Instances [7, 8, 9, 10, 11] were not killed in time',
'Exceeded maximum number of errors while killing instances']
+ self.assert_query(api, times=2)
def test_kill_job_with_message(self):
"""Test kill client-side API logic."""
@@ -470,6 +505,8 @@ class TestClientKillCommand(AuroraClientCommandTest):
patch('apache.aurora.client.cli.jobs.Job.create_context', return_value=mock_context),
patch('apache.aurora.client.cli.jobs.JobMonitor', return_value=mock_monitor)) as (_, m):
api = mock_context.get_api('west')
+ mock_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
api.kill_job.return_value = self.create_simple_success_response()
cmd = AuroraCommandLine()
message = 'Test message'
@@ -477,3 +514,4 @@ class TestClientKillCommand(AuroraClientCommandTest):
instances = [0]
self.assert_kill_calls(api, instances=instances, message=message)
+ self.assert_query(api)
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/sh/org/apache/aurora/e2e/http_example.py
----------------------------------------------------------------------
diff --git a/src/test/sh/org/apache/aurora/e2e/http_example.py b/src/test/sh/org/apache/aurora/e2e/http_example.py
index ba7d114..140cc75 100644
--- a/src/test/sh/org/apache/aurora/e2e/http_example.py
+++ b/src/test/sh/org/apache/aurora/e2e/http_example.py
@@ -14,7 +14,6 @@
from __future__ import print_function
from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
-from SocketServer import ThreadingMixIn
from sys import argv
from threading import Thread
@@ -41,6 +40,7 @@ def start_server(port, handler_class):
print('Listening on port %s' % port)
server.serve_forever()
+
request_thread = Thread(target=start_server, args=[int(argv[1]), RequestHandler])
health_thread = Thread(target=start_server, args=[int(argv[2]), HealthHandler])
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/sh/org/apache/aurora/e2e/partition_aware.aurora
----------------------------------------------------------------------
diff --git a/src/test/sh/org/apache/aurora/e2e/partition_aware.aurora b/src/test/sh/org/apache/aurora/e2e/partition_aware.aurora
index 7ea9fad..98732e4 100644
--- a/src/test/sh/org/apache/aurora/e2e/partition_aware.aurora
+++ b/src/test/sh/org/apache/aurora/e2e/partition_aware.aurora
@@ -1,3 +1,17 @@
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
# run the script
hello_world = Process(
name = 'hello_world',
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/sh/org/apache/aurora/e2e/sla_coordinator.py
----------------------------------------------------------------------
diff --git a/src/test/sh/org/apache/aurora/e2e/sla_coordinator.py b/src/test/sh/org/apache/aurora/e2e/sla_coordinator.py
new file mode 100644
index 0000000..24aa8f5
--- /dev/null
+++ b/src/test/sh/org/apache/aurora/e2e/sla_coordinator.py
@@ -0,0 +1,60 @@
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+from __future__ import print_function
+
+from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
+from threading import Thread
+from urlparse import urlparse, parse_qs
+
+import json
+import sys
+
+
+class RequestHandler(BaseHTTPRequestHandler):
+ def do_POST(self):
+ task = parse_qs(urlparse(self.path).query)['task'][0]
+ body = self.rfile.read(int(self.headers.getheader('content-length', 0)))
+
+ # Only allow draining instance-0
+ allow = True if task == 'devcluster/vagrant/test/coordinator/0' else False
+
+ self.send_response(200)
+ self.send_header('Content-Type', 'application/json')
+ self.end_headers()
+ self.wfile.write('{"drain":%s}' % allow)
+
+ print('Request received for {}'.format(task))
+ print(json.dumps(json.loads(body), indent=2, sort_keys=True))
+ print('Responded: {}'.format(allow))
+ sys.stdout.flush()
+
+
+def start_server(instance, port, handler_class):
+ # This service will act as its own SLA Coordinator.
+ # We need a static port for the coordinator service so the scheduler can communicate to it when
+ # checking SLA. We cannot have 2 tasks binding to the same port so instance-1 will bind to 8080
+ # as instance-0 is the one that will be acked for draining. instance-0 will bind to a random port.
+ if instance != 1:
+ port = 0
+
+ server = HTTPServer(('', port), handler_class)
+ print('Listening on port %s' % port)
+ sys.stdout.flush()
+ server.serve_forever()
+
+
+thread = Thread(target=start_server, args=[int(sys.argv[1]), 8080, RequestHandler])
+
+thread.start()
+thread.join()
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/sh/org/apache/aurora/e2e/sla_policy.aurora
----------------------------------------------------------------------
diff --git a/src/test/sh/org/apache/aurora/e2e/sla_policy.aurora b/src/test/sh/org/apache/aurora/e2e/sla_policy.aurora
new file mode 100644
index 0000000..d819778
--- /dev/null
+++ b/src/test/sh/org/apache/aurora/e2e/sla_policy.aurora
@@ -0,0 +1,61 @@
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+hello = Process(
+ name = 'hello',
+ cmdline = """
+ while true; do
+ echo hello world
+ sleep 10
+ done
+ """)
+
+copy_coordinator = Process(
+ name = 'copy_coordinator',
+ cmdline = 'cp /vagrant/src/test/sh/org/apache/aurora/e2e/sla_coordinator.py .'
+)
+
+run_coordinator = Process(
+ name = 'run_coordinator',
+ cmdline = 'python sla_coordinator.py {{mesos.instance}}')
+
+task = SequentialTask(
+ processes = [hello],
+ resources = Resources(cpu = 0.1, ram = 1*MB, disk = 8*MB))
+
+coordinator_task = SequentialTask(
+ processes = [copy_coordinator, run_coordinator],
+ resources = Resources(cpu = 0.1, ram = 1*MB, disk = 8*MB))
+
+service = Service(
+ task = task,
+ tier = 'preferred',
+ instances = 2,
+ cluster = 'devcluster',
+ role = 'vagrant',
+ environment = 'test')
+
+coordinator_service = Service(
+ task = coordinator_task,
+ tier = 'preferred',
+ instances = 2,
+ cluster = 'devcluster',
+ role = 'vagrant',
+ environment = 'test')
+
+jobs = [
+ service(name = 'count', sla_policy=CountSlaPolicy(count=1, duration_secs=60)),
+ service(name = 'percentage', sla_policy=PercentageSlaPolicy(percentage=50, duration_secs=60)),
+ coordinator_service(name = 'coordinator', sla_policy=CoordinatorSlaPolicy(coordinator_url='http://localhost:8080')),
+]
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/sh/org/apache/aurora/e2e/test_end_to_end.sh
----------------------------------------------------------------------
diff --git a/src/test/sh/org/apache/aurora/e2e/test_end_to_end.sh b/src/test/sh/org/apache/aurora/e2e/test_end_to_end.sh
index 888efe4..24f4448 100755
--- a/src/test/sh/org/apache/aurora/e2e/test_end_to_end.sh
+++ b/src/test/sh/org/apache/aurora/e2e/test_end_to_end.sh
@@ -33,11 +33,16 @@ _curl() { curl --silent --fail --retry 4 --retry-delay 10 "$@" ; }
tear_down() {
set +x # Disable command echo, as this makes it more difficult see which command failed.
- for job in http_example http_example_watch_secs http_example_revocable http_example_docker http_example_unified_appc http_example_unified_docker; do
- aurora update abort devcluster/vagrant/test/$job >/dev/null 2>&1 || true
- aurora job killall --no-batching devcluster/vagrant/test/$job >/dev/null 2>&1
+ local _jobs=$(aurora job list $TEST_CLUSTER/$TEST_ROLE| grep $TEST_ROLE)
+
+ for job in ${_jobs[@]}; do
+ aurora update abort $job >/dev/null 2>&1 || true
+ aurora job killall --no-batching $job >/dev/null 2>&1
done
+ aurora_admin set_quota $TEST_CLUSTER $TEST_ROLE 0 0m 0m
+ aurora_admin host_activate --hosts=$TEST_SLAVE_IP $TEST_CLUSTER
+
sudo mv /etc/aurora/clusters.json.old /etc/aurora/clusters.json >/dev/null 2>&1 || true
}
@@ -241,12 +246,50 @@ wait_until_task_status() {
_success=1
break
else
- sleep 1
+ sleep 20
+ fi
+ done
+
+ if [[ "$_success" -ne "1" ]]; then
+ echo "Task did not transition to $_expected_state within timeout."
+ exit 1
+ fi
+}
+
+assert_host_status() {
+ local _host=$1 _cluster=$2 _expected_state=$3
+
+ local _state=$(aurora_admin host_status --hosts=$_host $_cluster 2>&1 | tail -n1 | awk -F' ' '{print $6}')
+
+ if [[ $_state != $_expected_state ]]; then
+ echo "Expected host $_host to be in state $_expected_state, but found $_state"
+ exit 1
+ fi
+}
+
+wait_until_task_counts() {
+ # Poll the job, waiting for it to enter the target number of task counts
+ local _jobkey=$1 _expected_running=$2 _expected_pending=$3
+ local _num_running=0
+ local _num_pending=0
+ local _success=0
+
+ for i in $(seq 1 120); do
+ # || is so that we don't return an EXIT so that `trap collect_result` doesn't get triggered.
+ _num_running=$(aurora job status $_jobkey --write-json | jq -r ".[0].active[].status" | grep "RUNNING" | wc -l) || echo $?
+ _num_pending=$(aurora job status $_jobkey --write-json | jq -r ".[0].active[].status" | grep "PENDING" | wc -l) || echo $?
+
+ if [[ $_num_running == $_expected_running ]] && [[ $_num_pending == $_expected_pending ]]; then
+ _success=1
+ break
+ else
+ echo "Waiting for job $_jobkey to have $_expected_running RUNNING and $_expected_pending PENDING tasks."
+ sleep 20
fi
done
if [[ "$_success" -ne "1" ]]; then
- echo "Task did not transition to $_expected_state within two minutes."
+ echo "Job $_jobkey did not have $_expected_running RUNNING tasks and $_expected_pending PENDING tasks within timeout."
exit 1
fi
}
@@ -341,6 +384,73 @@ test_partition_awareness() {
aurora job killall $_delay_jobkey
}
+run_sla_aware_maintenance() {
+ local _config=$1
+ local _cluster=$2
+ local _jobkey=$3
+
+ aurora job create $_jobkey $_config --wait-until RUNNING
+
+ # assert the number of tasks, the job should have 2 RUNNING tasks
+ wait_until_task_counts $_jobkey 2 0
+
+ # check that the host starts with no maintenance mode
+ assert_host_status $TEST_SLAVE_IP $_cluster "NONE"
+
+ # trigger sla aware drain with default timeout of 2hr
+ # so, only allowed number (1 each) of tasks should drain for each job
+ aurora_admin sla_host_drain --hosts=$TEST_SLAVE_IP $_cluster
+
+ # force a scheduler restart and make sure that the maintenance request is still satisfied
+ sudo systemctl restart aurora-scheduler
+
+ # host must have maintenance mode set
+ assert_host_status $TEST_SLAVE_IP $_cluster "DRAINING"
+
+ # tasks get drained as allowed by the sla policy
+ wait_until_task_counts $_jobkey 1 1
+
+ # for coordinator sla check specific task states
+ if [[ $_jobkey == $TEST_JOB_COORDINATOR_SLA ]]; then
+ assert_task_status $_jobkey "0" PENDING
+ assert_task_status $_jobkey "1" RUNNING
+ fi
+
+ # host must have maintenance mode set and should be waiting in DRAINING
+ assert_host_status $TEST_SLAVE_IP $_cluster "DRAINING"
+
+ # force sla aware drain with zero timeout
+ aurora_admin sla_host_drain --force_drain_timeout=0s --hosts=$TEST_SLAVE_IP $_cluster
+
+ # tasks get drained as allowed by the sla policy
+ wait_until_task_counts $_jobkey 0 2
+
+ # activate host again
+ aurora_admin host_activate --hosts=$TEST_SLAVE_IP $_cluster
+
+ # assert the number of tasks the job should have 2 RUNNING tasks
+ wait_until_task_counts $_jobkey 2 0
+
+ # clean up
+ aurora job killall $_jobkey
+}
+
+test_sla_aware_maintenance() {
+ local _config=$1
+ local _cluster=$2
+ local _role=$3
+ local _count_jobkey=$4
+ local _percentage_jobkey=$5
+ local _coordinator_jobkey=$6
+
+ # add quota for each job (addl. for executor overhead) since only preferred jobs get sla policy
+ aurora_admin increase_quota $_cluster $_role 1.0 10m 50m
+
+ run_sla_aware_maintenance $_config $_cluster $_count_jobkey
+ run_sla_aware_maintenance $_config $_cluster $_percentage_jobkey
+ run_sla_aware_maintenance $_config $_cluster $_coordinator_jobkey
+}
+
test_announce() {
local _role=$1 _env=$2 _job=$3
@@ -744,6 +854,10 @@ TEST_PARTITION_AWARENESS_CONFIG_FILE=$TEST_ROOT/partition_aware.aurora
TEST_JOB_PA_DEFAULT=$TEST_CLUSTER/$TEST_ROLE/$TEST_ENV/partition_aware_default
TEST_JOB_PA_DISABLED=$TEST_CLUSTER/$TEST_ROLE/$TEST_ENV/partition_aware_disabled
TEST_JOB_PA_DELAY=$TEST_CLUSTER/$TEST_ROLE/$TEST_ENV/partition_aware_delay
+TEST_SLA_POLICY_CONFIG_FILE=$TEST_ROOT/sla_policy.aurora
+TEST_JOB_COUNT_SLA=$TEST_CLUSTER/$TEST_ROLE/$TEST_ENV/count
+TEST_JOB_PERCENTAGE_SLA=$TEST_CLUSTER/$TEST_ROLE/$TEST_ENV/percentage
+TEST_JOB_COORDINATOR_SLA=$TEST_CLUSTER/$TEST_ROLE/$TEST_ENV/coordinator
BASE_ARGS=(
$TEST_CLUSTER
@@ -792,6 +906,15 @@ TEST_PARTITION_AWARENESS_ARGS=(
$TEST_JOB_PA_DELAY
)
+TEST_SLA_AWARE_MAINTENANCE_ARGS=(
+ $TEST_SLA_POLICY_CONFIG_FILE
+ $TEST_CLUSTER
+ $TEST_ROLE
+ $TEST_JOB_COUNT_SLA
+ $TEST_JOB_PERCENTAGE_SLA
+ $TEST_JOB_COORDINATOR_SLA
+)
+
TEST_JOB_KILL_MESSAGE_ARGS=("${TEST_JOB_ARGS[@]}" "--message='Test message'")
trap collect_result EXIT
@@ -800,6 +923,8 @@ aurorabuild all
setup_ssh
setup_docker_registry
+test_sla_aware_maintenance "${TEST_SLA_AWARE_MAINTENANCE_ARGS[@]}"
+
test_partition_awareness "${TEST_PARTITION_AWARENESS_ARGS[@]}"
test_version
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/ui/src/main/js/components/TaskConfigSummary.js
----------------------------------------------------------------------
diff --git a/ui/src/main/js/components/TaskConfigSummary.js b/ui/src/main/js/components/TaskConfigSummary.js
index 64880f4..f03d44d 100644
--- a/ui/src/main/js/components/TaskConfigSummary.js
+++ b/ui/src/main/js/components/TaskConfigSummary.js
@@ -70,6 +70,46 @@ function PartitionPolicy({ config }) {
<td>{config.partitionPolicy.delaySecs}</td>
</tr>];
}
+
+function SlaPolicy({ config }) {
+ if (isNully(config.slaPolicy)) {
+ return null;
+ }
+
+ if (!isNully(config.slaPolicy.countSlaPolicy)) {
+ return [<tr>
+ <th rowSpan='2'>Count SLA Policy</th>
+ <td>count</td>
+ <td>{config.slaPolicy.countSlaPolicy.count}</td>
+ </tr>,
+ <tr>
+ <td>duration secs</td>
+ <td>{config.slaPolicy.countSlaPolicy.durationSecs}</td>
+ </tr>];
+ } else if (!isNully(config.slaPolicy.percentageSlaPolicy)) {
+ return [<tr>
+ <th rowSpan='2'>Percentage SLA Policy</th>
+ <td>precentage</td>
+ <td>{config.slaPolicy.percentageSlaPolicy.percentage + '%'}</td>
+ </tr>,
+ <tr>
+ <td>duration secs</td>
+ <td>{config.slaPolicy.percentageSlaPolicy.durationSecs}</td>
+ </tr>];
+ } else if (!isNully(config.slaPolicy.coordinatorSlaPolicy)) {
+ return [<tr>
+ <th rowSpan='4'>Coordinator SLA Policy</th>
+ <td>coordinator url</td>
+ <td>{config.slaPolicy.coordinatorSlaPolicy.coordinatorUrl}</td>
+ </tr>,
+ <tr>
+ <td>status key</td>
+ <td>{config.slaPolicy.coordinatorSlaPolicy.statusKey}</td>
+ </tr>];
+ }
+
+ return null;
+}
/* eslint-enable */
export function CronConfigSummary({ cronJob }) {
@@ -108,6 +148,7 @@ export function CronConfigSummary({ cronJob }) {
</tr>
<Metadata config={config} />
<PartitionPolicy config={config} />
+ <SlaPolicy config={config} />
<tr>
<th>Contact</th>
<td colSpan='2'>{config.contactEmail}</td>
@@ -136,6 +177,7 @@ export default function TaskConfigSummary({ config, instances }) {
</tr>
<Metadata config={config} />
<PartitionPolicy config={config} />
+ <SlaPolicy config={config} />
<tr>
<th>Contact</th>
<td colSpan='2'>{config.contactEmail}</td>
[4/4] aurora git commit: Enable `Tasks` to specify their own custom
maintenance SLA.
Posted by sa...@apache.org.
Enable `Tasks` to specify their own custom maintenance SLA.
`Tasks` can specify custom SLA requirements as part of
their `TaskConfig`. One of the new features is the ability
to specify an external coordinator that can ACK/NACK
maintenance requests for tasks. This will be hugely
beneficial for onboarding services that cannot satisfactorily
specify SLA in terms of running instances.
Maintenance requests are driven from the Scheduler to
improve management of nodes in the cluster.
Testing Done:
./build-support/jenkins/build.sh
./src/test/sh/org/apache/aurora/e2e/test_end_to_end.sh
Bugs closed: AURORA-1978
Reviewed at https://reviews.apache.org/r/66716/
Project: http://git-wip-us.apache.org/repos/asf/aurora/repo
Commit: http://git-wip-us.apache.org/repos/asf/aurora/commit/f2acf53f
Tree: http://git-wip-us.apache.org/repos/asf/aurora/tree/f2acf53f
Diff: http://git-wip-us.apache.org/repos/asf/aurora/diff/f2acf53f
Branch: refs/heads/master
Commit: f2acf53ff0d0cbb7b0c62f021cc55c9e45b32f9c
Parents: 34be631
Author: Santhosh Kumar Shanmugham <sa...@gmail.com>
Authored: Tue Jun 5 16:15:52 2018 -0700
Committer: Santhosh Kumar <ss...@twitter.com>
Committed: Tue Jun 5 16:15:52 2018 -0700
----------------------------------------------------------------------
RELEASE-NOTES.md | 32 +
.../thrift/org/apache/aurora/gen/api.thrift | 6 +
docs/README.md | 1 +
docs/features/sla-requirements.md | 181 +++
docs/operations/configuration.md | 55 +-
docs/reference/configuration.md | 34 +-
docs/reference/scheduler-configuration.md | 10 +
.../vagrant/systemd/aurora-scheduler.service | 6 +-
.../apache/aurora/scheduler/app/AppModule.java | 5 +
.../aurora/scheduler/config/CliOptions.java | 2 +
.../configuration/ConfigurationManager.java | 105 +-
.../maintenance/MaintenanceController.java | 470 +++++++
.../maintenance/MaintenanceModule.java | 78 ++
.../scheduler/mesos/MesosCallbackHandler.java | 2 +-
.../apache/aurora/scheduler/sla/SlaManager.java | 437 ++++++
.../apache/aurora/scheduler/sla/SlaModule.java | 57 +-
.../scheduler/state/MaintenanceController.java | 294 -----
.../aurora/scheduler/state/StateModule.java | 12 -
.../scheduler/thrift/ReadOnlySchedulerImpl.java | 10 +-
.../thrift/SchedulerThriftInterface.java | 24 +-
.../python/apache/aurora/admin/admin_util.py | 38 +
.../apache/aurora/admin/host_maintenance.py | 20 +
.../python/apache/aurora/admin/maintenance.py | 45 +
.../python/apache/aurora/client/api/__init__.py | 12 +
.../python/apache/aurora/client/cli/context.py | 16 +
.../python/apache/aurora/client/cli/jobs.py | 7 +
.../aurora/scheduler/app/SchedulerIT.java | 9 +
.../aurora/scheduler/base/TaskTestUtil.java | 4 +-
.../scheduler/config/CommandLineTest.java | 14 +-
.../configuration/ConfigurationManagerTest.java | 212 ++-
.../aurora/scheduler/cron/quartz/CronIT.java | 2 +-
.../scheduler/cron/quartz/QuartzTestUtil.java | 2 +-
.../MaintenanceControllerImplTest.java | 422 ++++++
.../mesos/MesosCallbackHandlerTest.java | 2 +-
.../aurora/scheduler/sla/SlaManagerTest.java | 1239 ++++++++++++++++++
.../aurora/scheduler/sla/SlaModuleTest.java | 15 +-
.../state/MaintenanceControllerImplTest.java | 264 ----
.../thrift/SchedulerThriftInterfaceTest.java | 58 +-
.../aurora/scheduler/thrift/ThriftIT.java | 12 +
.../thrift/aop/MockDecoratedThrift.java | 7 +
.../apache/aurora/admin/test_maintenance.py | 81 +-
src/test/python/apache/aurora/api_util.py | 3 +
.../aurora/client/api/test_scheduler_client.py | 9 +
.../python/apache/aurora/client/cli/test_add.py | 22 +-
.../apache/aurora/client/cli/test_kill.py | 52 +-
.../sh/org/apache/aurora/e2e/http_example.py | 2 +-
.../apache/aurora/e2e/partition_aware.aurora | 14 +
.../sh/org/apache/aurora/e2e/sla_coordinator.py | 60 +
.../sh/org/apache/aurora/e2e/sla_policy.aurora | 61 +
.../sh/org/apache/aurora/e2e/test_end_to_end.sh | 135 +-
ui/src/main/js/components/TaskConfigSummary.js | 42 +
51 files changed, 4058 insertions(+), 644 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/RELEASE-NOTES.md
----------------------------------------------------------------------
diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md
index 5e1f994..0ef75d6 100644
--- a/RELEASE-NOTES.md
+++ b/RELEASE-NOTES.md
@@ -1,3 +1,35 @@
+0.21.0
+======
+
+### New/updated:
+- Introduce ability for tasks to specify custom SLA requirements via the new `SlaPolicy` structs.
+ There are 3 different SLA Policies that are currently supported - `CountSlaPolicy`,
+ `PercentageSlaPolicy` and `CoordinatorSlaPolicy`. SLA policies based on count and percentage
+ express the required number of `RUNNING` instances as either a count or percentage in addition to
+ allowing the duration-window for which these requirements have to be satisfied. For applications
+ that need more control over how SLA is determined, a custom SLA calculator can be configured a.k.a
+ Coordinator. Any action that can affect SLA, will first check with the Coordinator before
+ proceeding.
+
+ **IMPORTANT: The storage changes required for this feature will make scheduler
+ snapshot backwards incompatible. Scheduler will be unable to read snapshot if rolled back to
+ previous version. If rollback is absolutely necessary, perform the following steps:**
+ 1. Stop all host maintenance requests via `aurora_admin host_activate`.
+ 2. Ensure a new snapshot is created by running `aurora_admin scheduler_snapshot <cluster>`
+ 3. Rollback to previous version
+
+ Note: The `Coordinator` interface required for the `CoordinatorSlaPolicy` is experimental at
+ this juncture and is bound to change in the future.
+
+### Deprecations and removals:
+
+- Deprecated the `aurora_admin host_drain` command used for maintenance. With this release the SLA
+ computations are moved to the scheduler and it is no longer required for the client to compute
+ SLAs and watch the drains. The scheduler persists any host maintenance request and performs
+ SLA-aware drain of the tasks, before marking the host as `DRAINED`. So maintenance requests
+ survive across scheduler fail-overs. Use the newly introduced `aurora_admin sla_host_drain`
+ to skip the SLA computations on the admin client.
+
0.20.0
======
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/api/src/main/thrift/org/apache/aurora/gen/api.thrift
----------------------------------------------------------------------
diff --git a/api/src/main/thrift/org/apache/aurora/gen/api.thrift b/api/src/main/thrift/org/apache/aurora/gen/api.thrift
index ff48000..54007c8 100644
--- a/api/src/main/thrift/org/apache/aurora/gen/api.thrift
+++ b/api/src/main/thrift/org/apache/aurora/gen/api.thrift
@@ -1244,6 +1244,12 @@ service AuroraAdmin extends AuroraSchedulerManager {
/** Set the given hosts back into serving mode. */
Response endMaintenance(1: Hosts hosts)
+ /**
+ * Ask scheduler to put hosts into DRAINING mode and move scheduled tasks off of the hosts
+ * such that its SLA requirements are satisfied. Use defaultSlaPolicy if it is not set for a task.
+ **/
+ Response slaDrainHosts(1: Hosts hosts, 2: SlaPolicy defaultSlaPolicy, 3: i64 timeoutSecs)
+
/** Start a storage snapshot and block until it completes. */
Response snapshot()
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/docs/README.md
----------------------------------------------------------------------
diff --git a/docs/README.md b/docs/README.md
index 166bf1c..8bf936d 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -28,6 +28,7 @@ Description of important Aurora features.
* [Services](features/services.md)
* [Service Discovery](features/service-discovery.md)
* [SLA Metrics](features/sla-metrics.md)
+ * [SLA Requirements](features/sla-requirements.md)
* [Webhooks](features/webhooks.md)
## Operators
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/docs/features/sla-requirements.md
----------------------------------------------------------------------
diff --git a/docs/features/sla-requirements.md b/docs/features/sla-requirements.md
new file mode 100644
index 0000000..555b174
--- /dev/null
+++ b/docs/features/sla-requirements.md
@@ -0,0 +1,181 @@
+SLA Requirements
+================
+
+- [Overview](#overview)
+- [Default SLA](#default-sla)
+- [Custom SLA](#custom-sla)
+ - [Count-based](#count-based)
+ - [Percentage-based](#percentage-based)
+ - [Coordinator-based](#coordinator-based)
+
+## Overview
+
+Aurora guarantees SLA requirements for jobs. These requirements limit the impact of cluster-wide
+maintenance operations on the jobs. For instance, when an operator upgrades
+the OS on all the Mesos agent machines, the tasks scheduled on them needs to be drained.
+By specifying the SLA requirements a job can make sure that it has enough instances to
+continue operating safely without incurring downtime.
+
+> SLA is defined as minimum number of active tasks required for a job every duration window.
+A task is active if it was in `RUNNING` state during the last duration window.
+
+There is a [default](#default-sla) SLA guarantee for
+[preferred](../features/multitenancy.md#configuration-tiers) tier jobs and it is also possible to
+specify [custom](#custom-sla) SLA requirements.
+
+## Default SLA
+
+Aurora guarantees a default SLA requirement for tasks in
+[preferred](../features/multitenancy.md#configuration-tiers) tier.
+
+> 95% of tasks in a job will be `active` for every 30 mins.
+
+
+## Custom SLA
+
+For jobs that require different SLA requirements, Aurora allows jobs to specify their own
+SLA requirements via the `SlaPolicies`. There are 3 different ways to express SLA requirements.
+
+### [Count-based](../reference/configuration.md#countslapolicy-objects)
+
+For jobs that need a minimum `number` of instances to be running all the time,
+[`CountSlaPolicy`](../reference/configuration.md#countslapolicy-objects)
+provides the ability to express the minimum number of required active instances (i.e. number of
+tasks that are `RUNNING` for at least `duration_secs`). For instance, if we have a
+`replicated-service` that has 3 instances and needs at least 2 instances every 30 minutes to be
+treated healthy, the SLA requirement can be expressed with a
+[`CountSlaPolicy`](../reference/configuration.md#countslapolicy-objects) like below,
+
+```python
+Job(
+ name = 'replicated-service',
+ role = 'www-data',
+ instances = 3,
+ sla_policy = CountSlaPolicy(
+ count = 2,
+ duration_secs = 1800
+ )
+ ...
+)
+```
+
+### [Percentage-based](../reference/configuration.md#percentageslapolicy-objects)
+
+For jobs that need a minimum `percentage` of instances to be running all the time,
+[`PercentageSlaPolicy`](../reference/configuration.md#percentageslapolicy-objects) provides the
+ability to express the minimum percentage of required active instances (i.e. percentage of tasks
+that are `RUNNING` for at least `duration_secs`). For instance, if we have a `webservice` that
+has 10000 instances for handling peak load and cannot have more than 0.1% of the instances down
+for every 1 hr, the SLA requirement can be expressed with a
+[`PercentageSlaPolicy`](../reference/configuration.md#percentageslapolicy-objects) like below,
+
+```python
+Job(
+ name = 'frontend-service',
+ role = 'www-data',
+ instances = 10000,
+ sla_policy = PercentageSlaPolicy(
+ percentage = 99.9,
+ duration_secs = 3600
+ )
+ ...
+)
+```
+
+### [Coordinator-based](../reference/configuration.md#coordinatorslapolicy-objects)
+
+When none of the above methods are enough to describe the SLA requirements for a job, then the SLA
+calculation can be off-loaded to a custom service called the `Coordinator`. The `Coordinator` needs
+to expose an endpoint that will be called to check if removal of a task will affect the SLA
+requirements for the job. This is useful to control the number of tasks that undergoes maintenance
+at a time, without affected the SLA for the application.
+
+Consider the example, where we have a `storage-service` stores 2 replicas of an object. Each replica
+is distributed across the instances, such that replicas are stored on different hosts. In addition
+a consistent-hash is used for distributing the data across the instances.
+
+When an instance needs to be drained (say for host-maintenance), we have to make sure that at least 1 of
+the 2 replicas remains available. In such a case, a `Coordinator` service can be used to maintain
+the SLA guarantees required for the job.
+
+The job can be configured with a
+[`CoordinatorSlaPolicy`](../reference/configuration.md#coordinatorslapolicy-objects) to specify the
+coordinator endpoint and the field in the response JSON that indicates if the SLA will be affected
+or not affected, when the task is removed.
+
+```python
+Job(
+ name = 'storage-service',
+ role = 'www-data',
+ sla_policy = CoordinatorSlaPolicy(
+ coordinator_url = 'http://coordinator.example.com',
+ status_key = 'drain'
+ )
+ ...
+)
+```
+
+
+#### Coordinator Interface [Experimental]
+
+When a [`CoordinatorSlaPolicy`](../reference/configuration.md#coordinatorslapolicy-objects) is
+specified for a job, any action that requires removing a task
+(such as drains) will be required to get approval from the `Coordinator` before proceeding. The
+coordinator service needs to expose a HTTP endpoint, that can take a `task-key` param
+(`<cluster>/<role>/<env>/<name>/<instance>`) and a json body describing the task
+details and return a response json that will contain the boolean status for allowing or disallowing
+the task's removal.
+
+##### Request:
+```javascript
+POST /
+ ?task=<cluster>/<role>/<env>/<name>/<instance>
+
+{
+ "assignedTask": {
+ "taskId": "taskA",
+ "slaveHost": "a",
+ "task": {
+ "job": {
+ "role": "role",
+ "environment": "devel",
+ "name": "job"
+ },
+ ...
+ },
+ "assignedPorts": {
+ "http": 1000
+ },
+ "instanceId": 1
+ ...
+ },
+ ...
+}
+```
+
+##### Response:
+```json
+{
+ "drain": true
+}
+```
+
+If Coordinator allows removal of the task, then the task’s
+[termination lifecycle](../reference/configuration.md#httplifecycleconfig-objects)
+is triggered. If Coordinator does not allow removal, then the request will be retried again in the
+future.
+
+#### Coordinator Actions
+
+Coordinator endpoint get its own lock and this is used to serializes calls to the Coordinator.
+It guarantees that only one concurrent request is sent to a coordinator endpoint. This allows
+coordinators to simply look the current state of the tasks to determine its SLA (without having
+to worry about in-flight and pending requests). However if there are multiple coordinators,
+maintenance can be done in parallel across all the coordinators.
+
+_Note: Single concurrent request to a coordinator endpoint does not translate as exactly-once
+guarantee. The coordinator must be able to handle duplicate drain
+requests for the same task._
+
+
+
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/docs/operations/configuration.md
----------------------------------------------------------------------
diff --git a/docs/operations/configuration.md b/docs/operations/configuration.md
index 85a6fab..4890c09 100644
--- a/docs/operations/configuration.md
+++ b/docs/operations/configuration.md
@@ -312,19 +312,19 @@ increased).
To enable this in the Scheduler, you can set the following options:
- --enable_update_affinity=true
- --update_affinity_reservation_hold_time=3mins
+ -enable_update_affinity=true
+ -update_affinity_reservation_hold_time=3mins
You will need to tune the hold time to match the behavior you see in your cluster. If you have extremely
high update throughput, you might have to extend it as processing updates could easily add significant
delays between scheduling attempts. You may also have to tune scheduling parameters to achieve the
throughput you need in your cluster. Some relevant settings (with defaults) are:
- --max_schedule_attempts_per_sec=40
- --initial_schedule_penalty=1secs
- --max_schedule_penalty=1mins
- --scheduling_max_batch_size=3
- --max_tasks_per_schedule_attempt=5
+ -max_schedule_attempts_per_sec=40
+ -initial_schedule_penalty=1secs
+ -max_schedule_penalty=1mins
+ -scheduling_max_batch_size=3
+ -max_tasks_per_schedule_attempt=5
There are metrics exposed by the Scheduler which can provide guidance on where the bottleneck is.
Example metrics to look at:
@@ -337,3 +337,44 @@ Example metrics to look at:
Most likely you'll run into limits with the number of update instances that can be processed per minute
before you run into any other limits. So if your total work done per minute starts to exceed 2k instances,
you may need to extend the update_affinity_reservation_hold_time.
+
+## Cluster Maintenance
+
+Aurora performs maintenance related task drains. One of the scheduler options that can control
+how often the scheduler polls for maintenance work can be controlled via,
+
+ -host_maintenance_polling_interval=1min
+
+## Enforcing SLA limitations
+
+Since tasks can specify their own `SLAPolicy`, the cluster needs to limit these SLA requirements.
+Too aggressive a requirement can permanently block any type of maintenance work
+(ex: OS/Kernel/Security upgrades) on a host and hold it hostage.
+
+An operator can control the limits for SLA requirements via these scheduler configuration options:
+
+ -max_sla_duration_secs=2hrs
+ -min_required_instances_for_sla_check=20
+
+_Note: These limits only apply for `CountSlaPolicy` and `PercentageSlaPolicy`._
+
+### Limiting Coordinator SLA
+
+With `CoordinatorSlaPolicy` the SLA calculation is off-loaded to an external HTTP service. Some
+relevant scheduler configuration options are,
+
+ -sla_coordinator_timeout=1min
+ -max_parallel_coordinated_maintenance=10
+
+Since handing off the SLA calculation to an external service can potentially block maintenance
+on hosts for an indefinite amount of time (either due to a mis-configured coordinator or due to
+a valid degraded service). In those situations the following metrics will be helpful to identify the
+offending tasks.
+
+ sla_coordinator_user_errors_* (counter tracking number of times the coordinator for the task
+ returned a bad response.)
+ sla_coordinator_errors_* (counter tracking number of times the scheduler was not able
+ to communicate with the coordinator of the task.)
+ sla_coordinator_lock_starvation_* (counter tracking number of times the scheduler was not able to
+ get the lock for the coordinator of the task.)
+
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/docs/reference/configuration.md
----------------------------------------------------------------------
diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md
index d4b869b..acab4c5 100644
--- a/docs/reference/configuration.md
+++ b/docs/reference/configuration.md
@@ -23,6 +23,7 @@ configuration design.
- [Announcer Objects](#announcer-objects)
- [Container Objects](#container)
- [LifecycleConfig Objects](#lifecycleconfig-objects)
+ - [SlaPolicy Objects](#slapolicy-objects)
- [Specifying Scheduling Constraints](#specifying-scheduling-constraints)
- [Template Namespaces](#template-namespaces)
- [mesos Namespace](#mesos-namespace)
@@ -343,7 +344,7 @@ Job Schema
```contact``` | String | Best email address to reach the owner of the job. For production jobs, this is usually a team mailing list.
```instances```| Integer | Number of instances (sometimes referred to as replicas or shards) of the task to create. (Default: 1)
```cron_schedule``` | String | Cron schedule in cron format. May only be used with non-service jobs. See [Cron Jobs](../features/cron-jobs.md) for more information. Default: None (not a cron job.)
- ```cron_collision_policy``` | String | Policy to use when a cron job is triggered while a previous run is still active. KILL_EXISTING Kill the previous run, and schedule the new run CANCEL_NEW Let the previous run continue, and cancel the new run. (Default: KILL_EXISTING)
+ ```cron_collision_policy``` | String | Policy to use when a cron job is triggered while a previous run is still active. KILL\_EXISTING Kill the previous run, and schedule the new run CANCEL\_NEW Let the previous run continue, and cancel the new run. (Default: KILL_EXISTING)
```update_config``` | ```UpdateConfig``` object | Parameters for controlling the rate and policy of rolling updates.
```constraints``` | dict | Scheduling constraints for the tasks. See the section on the [constraint specification language](#specifying-scheduling-constraints)
```service``` | Boolean | If True, restart tasks regardless of success or failure. (Default: False)
@@ -359,6 +360,7 @@ Job Schema
```partition_policy``` | ```PartitionPolicy``` object | An optional partition policy that allows job owners to define how to handle partitions for running tasks (in partition-aware Aurora clusters)
```metadata``` | list of ```Metadata``` objects | list of ```Metadata``` objects for user's customized metadata information.
```executor_config``` | ```ExecutorConfig``` object | Allows choosing an alternative executor defined in `custom_executor_config` to be used instead of Thermos. Tasks will be launched with Thermos as the executor by default. See [Custom Executors](../features/custom-executors.md) for more info.
+ ```sla_policy``` | Choice of ```CountSlaPolicy```, ```PercentageSlaPolicy``` or ```CoordinatorSlaPolicy``` object | An optional SLA policy that allows job owners to describe the SLA requirements for the job. See [SlaPolicy Objects](#slapolicy-objects) for more information.
### UpdateConfig Objects
@@ -564,7 +566,7 @@ See [Docker Command Line Reference](https://docs.docker.com/reference/commandlin
```graceful_shutdown_wait_secs``` | Integer | The amount of time (in seconds) to wait after hitting the ```graceful_shutdown_endpoint``` before proceeding with the [task termination lifecycle](https://aurora.apache.org/documentation/latest/reference/task-lifecycle/#forceful-termination-killing-restarting). (Default: 5)
```shutdown_wait_secs``` | Integer | The amount of time (in seconds) to wait after hitting the ```shutdown_endpoint``` before proceeding with the [task termination lifecycle](https://aurora.apache.org/documentation/latest/reference/task-lifecycle/#forceful-termination-killing-restarting). (Default: 5)
-#### graceful_shutdown_endpoint
+#### graceful\_shutdown\_endpoint
If the Job is listening on the port as specified by the HttpLifecycleConfig
(default: `health`), a HTTP POST request will be sent over localhost to this
@@ -581,6 +583,34 @@ does not shut down on its own after `shutdown_wait_secs` seconds, it will be
forcefully killed.
+### SlaPolicy Objects
+
+Configuration for specifying custom [SLA requirements](../features/sla-requirements.md) for a job. There are 3 supported SLA policies
+namely, [`CountSlaPolicy`](#countslapolicy-objects), [`PercentageSlaPolicy`](#percentageslapolicy-objects) and [`CoordinatorSlaPolicy`](#coordinatorslapolicy-objects).
+
+
+### CountSlaPolicy Objects
+
+ param | type | description
+ ----- | :----: | -----------
+ ```count``` | Integer | The number of active instances required every `durationSecs`.
+ ```duration_secs``` | Integer | Minimum time duration a task needs to be `RUNNING` to be treated as active.
+
+### PercentageSlaPolicy Objects
+
+ param | type | description
+ ----- | :----: | -----------
+ ```percentage``` | Float | The percentage of active instances required every `durationSecs`.
+ ```duration_secs``` | Integer | Minimum time duration a task needs to be `RUNNING` to be treated as active.
+
+### CoordinatorSlaPolicy Objects
+
+ param | type | description
+ ----- | :----: | -----------
+ ```coordinator_url``` | String | The URL to the [Coordinator](../features/sla-requirements.md#coordinator) service to be contacted before performing SLA affecting actions (job updates, host drains etc).
+ ```status_key``` | String | The field in the Coordinator response that indicates the SLA status for working on the task. (Default: `drain`)
+
+
Specifying Scheduling Constraints
=================================
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/docs/reference/scheduler-configuration.md
----------------------------------------------------------------------
diff --git a/docs/reference/scheduler-configuration.md b/docs/reference/scheduler-configuration.md
index a659cfa..805e516 100644
--- a/docs/reference/scheduler-configuration.md
+++ b/docs/reference/scheduler-configuration.md
@@ -106,6 +106,8 @@ Optional flags:
Minimum guaranteed time for task history retention before any pruning is attempted.
-history_prune_threshold (default (2, days))
Time after which the scheduler will prune terminated task history.
+-host_maintenance_polling_interval (default (1, minute))
+ Interval between polling for pending host maintenance requests.
-hostname
The hostname to advertise in ZooKeeper instead of the locally-resolved hostname.
-http_authentication_mechanism (default NONE)
@@ -134,6 +136,8 @@ Optional flags:
Maximum delay between attempts to schedule a flapping task.
-max_leading_duration (default (1, days))
After leading for this duration, the scheduler should commit suicide.
+-max_parallel_coordinated_maintenance (default 10)
+ Maximum number of coordinators that can be contacted in parallel.
-max_registration_delay (default (1, mins))
Max allowable delay to allow the driver to register before aborting
-max_reschedule_task_delay_on_startup (default (30, secs))
@@ -144,6 +148,8 @@ Optional flags:
Maximum number of scheduling attempts to make per second.
-max_schedule_penalty (default (1, mins))
Maximum delay between attempts to schedule a PENDING tasks.
+-max_sla_duration_secs (default (2, hrs))
+ Maximum duration window for which SLA requirements are to be satisfied. This does not apply to jobs that have a CoordinatorSlaPolicy.
-max_status_update_batch_size (default 1000) [must be > 0]
The maximum number of status updates that can be processed in a batch.
-max_task_event_batch_size (default 300) [must be > 0]
@@ -156,6 +162,8 @@ Optional flags:
Upper limit on the number of failures allowed during a job update. This helps cap potentially unbounded entries into storage.
-min_offer_hold_time (default (5, mins))
Minimum amount of time to hold a resource offer before declining.
+-min_required_instances_for_sla_check (default 20)
+ Minimum number of instances required for a job to be eligible for SLA check. This does not apply to jobs that have a CoordinatorSlaPolicy.
-native_log_election_retries (default 20)
The maximum number of attempts to obtain a new log writer.
-native_log_election_timeout (default (15, secs))
@@ -214,6 +222,8 @@ Optional flags:
Path to shiro.ini for authentication and authorization configuration.
-shiro_realm_modules (default [class org.apache.aurora.scheduler.http.api.security.IniShiroRealmModule])
Guice modules for configuring Shiro Realms.
+-sla_coordinator_timeout (default (1, min)) [must be > 0]
+ Timeout interval for communicating with Coordinator.
-sla_non_prod_metrics (default [])
Metric categories collected for non production tasks.
-sla_prod_metrics (default [JOB_UPTIMES, PLATFORM_UPTIME, MEDIANS])
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/examples/vagrant/systemd/aurora-scheduler.service
----------------------------------------------------------------------
diff --git a/examples/vagrant/systemd/aurora-scheduler.service b/examples/vagrant/systemd/aurora-scheduler.service
index 57e4bba..fa76ad0 100644
--- a/examples/vagrant/systemd/aurora-scheduler.service
+++ b/examples/vagrant/systemd/aurora-scheduler.service
@@ -31,6 +31,8 @@ Environment="JAVA_OPTS='-Djava.library.path=/usr/lib \
-Xdebug \
-Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=5005'"
WorkingDirectory=/home/vagrant/aurora/dist/install/aurora-scheduler
+
+# min_offer_hold_time is set to 1mins to shorten time to update attributes on offers. AURORA-1985.
ExecStart=/home/vagrant/aurora/dist/install/aurora-scheduler/bin/aurora-scheduler \
-cluster_name=devcluster \
-hostname=aurora.local \
@@ -58,7 +60,9 @@ ExecStart=/home/vagrant/aurora/dist/install/aurora-scheduler/bin/aurora-schedule
-allow_container_volumes=true \
-offer_filter_duration=0secs \
-mesos_driver=V1_DRIVER \
- -unavailability_threshold=1mins
+ -unavailability_threshold=1mins \
+ -min_required_instances_for_sla_check=2 \
+ -min_offer_hold_time=1mins
User=root
Group=root
Restart=always
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/java/org/apache/aurora/scheduler/app/AppModule.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/app/AppModule.java b/src/main/java/org/apache/aurora/scheduler/app/AppModule.java
index ffc0744..edc13d4 100644
--- a/src/main/java/org/apache/aurora/scheduler/app/AppModule.java
+++ b/src/main/java/org/apache/aurora/scheduler/app/AppModule.java
@@ -26,6 +26,7 @@ import com.google.inject.AbstractModule;
import org.apache.aurora.GuiceUtils;
import org.apache.aurora.common.inject.TimedInterceptor;
+import org.apache.aurora.common.quantity.Time;
import org.apache.aurora.common.stats.Stats;
import org.apache.aurora.common.stats.StatsProvider;
import org.apache.aurora.common.util.Clock;
@@ -46,6 +47,7 @@ import org.apache.aurora.scheduler.events.PubsubEventModule;
import org.apache.aurora.scheduler.filter.SchedulingFilter;
import org.apache.aurora.scheduler.filter.SchedulingFilterImpl;
import org.apache.aurora.scheduler.http.JettyServerModule;
+import org.apache.aurora.scheduler.maintenance.MaintenanceModule;
import org.apache.aurora.scheduler.mesos.SchedulerDriverModule;
import org.apache.aurora.scheduler.metadata.MetadataModule;
import org.apache.aurora.scheduler.offers.OfferManagerModule;
@@ -146,6 +148,8 @@ public class AppModule extends AbstractModule {
opts.main.allowGpuResource,
opts.app.enableMesosFetcher,
opts.app.allowContainerVolumes,
+ opts.sla.minRequiredInstances,
+ opts.sla.maxSlaDuration.as(Time.SECONDS),
opts.app.allowedJobEnvironments),
opts.main.driverImpl,
opts);
@@ -190,6 +194,7 @@ public class AppModule extends AbstractModule {
install(new StateModule(options));
install(new SlaModule(options.sla));
install(new UpdaterModule(options.updater));
+ install(new MaintenanceModule(options.maintenance));
bind(StatsProvider.class).toInstance(Stats.STATS_PROVIDER);
}
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/java/org/apache/aurora/scheduler/config/CliOptions.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/config/CliOptions.java b/src/main/java/org/apache/aurora/scheduler/config/CliOptions.java
index a2fb039..2dbd535 100644
--- a/src/main/java/org/apache/aurora/scheduler/config/CliOptions.java
+++ b/src/main/java/org/apache/aurora/scheduler/config/CliOptions.java
@@ -35,6 +35,7 @@ import org.apache.aurora.scheduler.http.api.security.HttpSecurityModule;
import org.apache.aurora.scheduler.http.api.security.IniShiroRealmModule;
import org.apache.aurora.scheduler.http.api.security.Kerberos5ShiroRealmModule;
import org.apache.aurora.scheduler.log.mesos.MesosLogStreamModule;
+import org.apache.aurora.scheduler.maintenance.MaintenanceModule;
import org.apache.aurora.scheduler.mesos.CommandLineDriverSettingsModule;
import org.apache.aurora.scheduler.offers.OfferManagerModule;
import org.apache.aurora.scheduler.preemptor.PreemptorModule;
@@ -86,6 +87,7 @@ public class CliOptions {
public final StatsModule.Options stats = new StatsModule.Options();
public final CronModule.Options cron = new CronModule.Options();
public final ResourceSettings resourceSettings = new ResourceSettings();
+ public final MaintenanceModule.Options maintenance = new MaintenanceModule.Options();
final List<Object> custom;
public CliOptions() {
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/java/org/apache/aurora/scheduler/configuration/ConfigurationManager.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/configuration/ConfigurationManager.java b/src/main/java/org/apache/aurora/scheduler/configuration/ConfigurationManager.java
index 4073229..2fd1746 100644
--- a/src/main/java/org/apache/aurora/scheduler/configuration/ConfigurationManager.java
+++ b/src/main/java/org/apache/aurora/scheduler/configuration/ConfigurationManager.java
@@ -17,7 +17,6 @@ import java.util.List;
import java.util.Optional;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
-
import javax.annotation.Nullable;
import javax.inject.Inject;
@@ -30,8 +29,11 @@ import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import org.apache.aurora.gen.Container;
+import org.apache.aurora.gen.CountSlaPolicy;
import org.apache.aurora.gen.DockerParameter;
import org.apache.aurora.gen.JobConfiguration;
+import org.apache.aurora.gen.PercentageSlaPolicy;
+import org.apache.aurora.gen.SlaPolicy;
import org.apache.aurora.gen.TaskConfig;
import org.apache.aurora.gen.TaskConstraint;
import org.apache.aurora.scheduler.TierManager;
@@ -99,6 +101,8 @@ public class ConfigurationManager {
private final boolean enableMesosFetcher;
private final boolean allowContainerVolumes;
private final Pattern allowedJobEnvironments;
+ private final int minRequiredInstances;
+ private final long maxSlaDurationSecs;
public ConfigurationManagerSettings(
ImmutableSet<Container._Fields> allowedContainerTypes,
@@ -108,6 +112,8 @@ public class ConfigurationManager {
boolean allowGpuResource,
boolean enableMesosFetcher,
boolean allowContainerVolumes,
+ int minRequiredInstances,
+ long maxSlaDurationSecs,
String allowedJobEnvironment) {
this.allowedContainerTypes = requireNonNull(allowedContainerTypes);
@@ -118,6 +124,8 @@ public class ConfigurationManager {
this.enableMesosFetcher = enableMesosFetcher;
this.allowContainerVolumes = allowContainerVolumes;
this.allowedJobEnvironments = Pattern.compile(requireNonNull(allowedJobEnvironment));
+ this.minRequiredInstances = minRequiredInstances;
+ this.maxSlaDurationSecs = maxSlaDurationSecs;
}
}
@@ -196,7 +204,9 @@ public class ConfigurationManager {
}
builder.setTaskConfig(
- validateAndPopulate(ITaskConfig.build(builder.getTaskConfig())).newBuilder());
+ validateAndPopulate(
+ ITaskConfig.build(builder.getTaskConfig()),
+ builder.getInstanceCount()).newBuilder());
// Only one of [service=true, cron_schedule] may be set.
if (!Strings.isNullOrEmpty(job.getCronSchedule()) && builder.getTaskConfig().isIsService()) {
@@ -207,6 +217,89 @@ public class ConfigurationManager {
return IJobConfiguration.build(builder);
}
+ /**
+ * Validates the {@link TaskConfig}'s {@link SlaPolicy}, if any.
+ *
+ * A valid {@link SlaPolicy} is one that allows at least 1 instance to be drained for a job.
+ *
+ * @param config {@link TaskConfig} to be validated.
+ * @param instanceCount number of instances in the job.
+ * @throws TaskDescriptionException thrown when {@link SlaPolicy} is not valid.
+ */
+ private void validateSlaPolicy(
+ TaskConfig config,
+ int instanceCount) throws TaskDescriptionException {
+
+ if (config.isSetSlaPolicy()) {
+ if (tierManager.getTier(ITaskConfig.build(config)).isRevocable()
+ || tierManager.getTier(ITaskConfig.build(config)).isPreemptible()) {
+ throw new TaskDescriptionException(String.format(
+ "Tier '%s' does not support SlaPolicy.",
+ config.getTier()));
+ }
+
+ SlaPolicy slaPolicy = config.getSlaPolicy();
+ if (!slaPolicy.isSetCoordinatorSlaPolicy()
+ && instanceCount < settings.minRequiredInstances) {
+ throw new TaskDescriptionException(String.format(
+ "Job with fewer than %d instances cannot have Percentage/Count SlaPolicy.",
+ settings.minRequiredInstances));
+ }
+
+ if (slaPolicy.isSetCountSlaPolicy()) {
+ validateCountSlaPolicy(instanceCount, slaPolicy.getCountSlaPolicy());
+ }
+
+ if (slaPolicy.isSetPercentageSlaPolicy()) {
+ validatePercentageSlaPolicy(instanceCount, slaPolicy.getPercentageSlaPolicy());
+ }
+ }
+ }
+
+ private void validatePercentageSlaPolicy(
+ int instanceCount,
+ PercentageSlaPolicy slaPolicy) throws TaskDescriptionException {
+ if (slaPolicy.isSetPercentage()) {
+ double percentage = slaPolicy.getPercentage() / 100.0;
+ if (instanceCount - instanceCount * percentage < 1) {
+ throw new TaskDescriptionException(String.format(
+ "Current PercentageSlaPolicy: percentage=%f will not allow any instances to be killed. "
+ + "Must be less than %f.",
+ slaPolicy.getPercentage(),
+ ((double) (instanceCount - 1)) / instanceCount * 100.0));
+ }
+ }
+
+ if (slaPolicy.isSetDurationSecs()
+ && slaPolicy.getDurationSecs() > settings.maxSlaDurationSecs) {
+ throw new TaskDescriptionException(String.format(
+ "PercentageSlaPolicy: durationSecs=%d must be less than cluster-wide maximum of %d secs.",
+ slaPolicy.getDurationSecs(),
+ settings.maxSlaDurationSecs));
+ }
+ }
+
+ private void validateCountSlaPolicy(
+ int instanceCount,
+ CountSlaPolicy slaPolicy) throws TaskDescriptionException {
+ if (slaPolicy.isSetCount()
+ && instanceCount - slaPolicy.getCount() < 1) {
+ throw new TaskDescriptionException(String.format(
+ "Current CountSlaPolicy: count=%d will not allow any instances to be killed. "
+ + "Must be less than instanceCount=%d.",
+ slaPolicy.getCount(),
+ instanceCount));
+ }
+
+ if (slaPolicy.isSetDurationSecs()
+ && slaPolicy.getDurationSecs() > settings.maxSlaDurationSecs) {
+ throw new TaskDescriptionException(String.format(
+ "CountSlaPolicy: durationSecs=%d must be less than cluster-wide maximum of %d secs.",
+ slaPolicy.getDurationSecs(),
+ settings.maxSlaDurationSecs));
+ }
+ }
+
@VisibleForTesting
static final String NO_DOCKER_PARAMETERS =
"This scheduler is configured to disallow Docker parameters.";
@@ -236,10 +329,14 @@ public class ConfigurationManager {
*
*
* @param config Task config to validate and populate.
+ * @param instanceCount The number of instances in the job.
* @return A reference to the modified {@code config} (for chaining).
* @throws TaskDescriptionException If the task is invalid.
*/
- public ITaskConfig validateAndPopulate(ITaskConfig config) throws TaskDescriptionException {
+ public ITaskConfig validateAndPopulate(
+ ITaskConfig config,
+ int instanceCount) throws TaskDescriptionException {
+
TaskConfig builder = config.newBuilder();
if (config.isSetTier() && !UserProvidedStrings.isGoodIdentifier(config.getTier())) {
@@ -374,6 +471,8 @@ public class ConfigurationManager {
}
}
+ validateSlaPolicy(builder, instanceCount);
+
maybeFillLinks(builder);
return ITaskConfig.build(builder);
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/java/org/apache/aurora/scheduler/maintenance/MaintenanceController.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/maintenance/MaintenanceController.java b/src/main/java/org/apache/aurora/scheduler/maintenance/MaintenanceController.java
new file mode 100644
index 0000000..dd2462d
--- /dev/null
+++ b/src/main/java/org/apache/aurora/scheduler/maintenance/MaintenanceController.java
@@ -0,0 +1,470 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.maintenance;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
+import javax.inject.Inject;
+import javax.inject.Qualifier;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.base.Predicates;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Sets;
+import com.google.common.eventbus.Subscribe;
+import com.google.common.util.concurrent.AbstractScheduledService;
+
+import org.apache.aurora.common.inject.TimedInterceptor.Timed;
+import org.apache.aurora.common.quantity.Amount;
+import org.apache.aurora.common.quantity.Time;
+import org.apache.aurora.common.stats.StatsProvider;
+import org.apache.aurora.gen.HostMaintenanceRequest;
+import org.apache.aurora.gen.HostStatus;
+import org.apache.aurora.gen.MaintenanceMode;
+import org.apache.aurora.gen.PercentageSlaPolicy;
+import org.apache.aurora.gen.ScheduleStatus;
+import org.apache.aurora.gen.SlaPolicy;
+import org.apache.aurora.scheduler.BatchWorker;
+import org.apache.aurora.scheduler.SchedulerModule.TaskEventBatchWorker;
+import org.apache.aurora.scheduler.base.Query;
+import org.apache.aurora.scheduler.base.Tasks;
+import org.apache.aurora.scheduler.config.types.TimeAmount;
+import org.apache.aurora.scheduler.events.PubsubEvent.EventSubscriber;
+import org.apache.aurora.scheduler.events.PubsubEvent.TaskStateChange;
+import org.apache.aurora.scheduler.sla.SlaManager;
+import org.apache.aurora.scheduler.state.StateManager;
+import org.apache.aurora.scheduler.storage.AttributeStore;
+import org.apache.aurora.scheduler.storage.Storage;
+import org.apache.aurora.scheduler.storage.Storage.MutableStoreProvider;
+import org.apache.aurora.scheduler.storage.Storage.StoreProvider;
+import org.apache.aurora.scheduler.storage.entities.IHostAttributes;
+import org.apache.aurora.scheduler.storage.entities.IHostMaintenanceRequest;
+import org.apache.aurora.scheduler.storage.entities.IHostStatus;
+import org.apache.aurora.scheduler.storage.entities.IScheduledTask;
+import org.apache.aurora.scheduler.storage.entities.ISlaPolicy;
+import org.apache.mesos.v1.Protos;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static java.lang.annotation.ElementType.FIELD;
+import static java.lang.annotation.ElementType.METHOD;
+import static java.lang.annotation.ElementType.PARAMETER;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+import static java.util.Objects.requireNonNull;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import static org.apache.aurora.gen.MaintenanceMode.DRAINED;
+import static org.apache.aurora.gen.MaintenanceMode.DRAINING;
+
+/**
+ * Logic that puts hosts into maintenance mode, and triggers draining of hosts upon request.
+ * All state-changing functions return their results. Additionally, all state-changing functions
+ * will ignore requests to change state of unknown hosts and subsequently omit these hosts from
+ * return values.
+ */
+public interface MaintenanceController {
+
+ /**
+ * Places hosts in maintenance mode.
+ * Hosts in maintenance mode are less-preferred for scheduling.
+ * No change will be made for hosts that are not recognized, and unrecognized hosts will not be
+ * included in the result.
+ *
+ * @param hosts Hosts to put into maintenance mode.
+ * @return The adjusted state of the hosts.
+ */
+ Set<IHostStatus> startMaintenance(Set<String> hosts);
+
+ /**
+ * Initiate a drain of all active tasks on {@code hosts}.
+ *
+ * @param hosts Hosts to drain.
+ * @return The adjusted state of the hosts. Hosts without any active tasks will be immediately
+ * moved to DRAINED.
+ */
+ Set<IHostStatus> drain(Set<String> hosts);
+
+ /**
+ * Initiate an SLA-aware drain of all active tasks on {@code hosts}.
+ *
+ * @param hosts Hosts to drain.
+ * @param defaultSlaPolicy SlaPolicy to use if a task does not have an SlaPolicy.
+ * @param timeoutSecs Interval after which tasks will be forcefully drained without checking SLA.
+ * @return The adjusted state of the hosts. Hosts without any active tasks will be immediately
+ * moved to DRAINED.
+ */
+ Set<IHostStatus> slaDrain(Set<String> hosts, SlaPolicy defaultSlaPolicy, long timeoutSecs);
+
+ /**
+ * Drain tasks defined by the inverse offer.
+ * This method doesn't set any host attributes.
+ *
+ * @param inverseOffer the inverse offer to use.
+ */
+ void drainForInverseOffer(Protos.InverseOffer inverseOffer);
+
+ /**
+ * Fetches the current maintenance mode of {$code host}.
+ *
+ * @param host Host to fetch state for.
+ * @return Maintenance mode of host, {@link MaintenanceMode#NONE} if the host is not known.
+ */
+ MaintenanceMode getMode(String host);
+
+ /**
+ * Fetches the current state of {@code hosts}.
+ *
+ * @param hosts Hosts to fetch state for.
+ * @return The state of the hosts.
+ */
+ Set<IHostStatus> getStatus(Set<String> hosts);
+
+ /**
+ * Moves {@code hosts} out of maintenance mode, returning them to mode NONE.
+ *
+ * @param hosts Hosts to move out of maintenance mode.
+ * @return The adjusted state of the hosts.
+ */
+ Set<IHostStatus> endMaintenance(Set<String> hosts);
+
+ /**
+ * Records the maintenance requests for hosts and drains any active tasks from the host
+ * asynchronously.
+ *
+ * Tasks are drained iff it will satisfy the required SLA for the task. Task's SLA is either the
+ * {@link SlaPolicy} configured as part of the TaskConfig or the default {@link SlaPolicy}
+ * specified as part of the maintenance request. If neither then the task is drained immediately.
+ *
+ * In order to avoid tasks from blocking maintenance perpetually each maintenance request has a
+ * timeout after which all tasks forcefully drained.
+ */
+ class MaintenanceControllerImpl
+ extends AbstractScheduledService implements MaintenanceController, EventSubscriber {
+ private static final Logger LOG = LoggerFactory.getLogger(MaintenanceControllerImpl.class);
+
+ @VisibleForTesting
+ @Qualifier
+ @Target({ FIELD, PARAMETER, METHOD }) @Retention(RUNTIME)
+ public @interface PollingInterval { }
+
+ @VisibleForTesting
+ static final String DRAINING_MESSAGE = "Draining machine for maintenance.";
+
+ private static final String MISSING_MAINTENANCE_REQUEST = "missing_maintenance_request";
+ private static final SlaPolicy ZERO_PERCENT_SLA = SlaPolicy.percentageSlaPolicy(
+ new PercentageSlaPolicy()
+ .setPercentage(0)
+ .setDurationSecs(0));
+
+ private final Storage storage;
+ private final Amount<Long, Time> pollingInterval;
+ private final TaskEventBatchWorker batchWorker;
+ private final SlaManager slaManager;
+ private final StateManager stateManager;
+
+ private final AtomicLong missingMaintenanceCounter;
+
+ @Inject
+ public MaintenanceControllerImpl(
+ Storage storage,
+ @PollingInterval Amount<Long, Time> pollingInterval,
+ TaskEventBatchWorker batchWorker,
+ SlaManager slaManager,
+ StateManager stateManager,
+ StatsProvider statsProvider) {
+
+ this.storage = requireNonNull(storage);
+ this.pollingInterval = checkNotNull(pollingInterval);
+ this.batchWorker = requireNonNull(batchWorker);
+ this.slaManager = requireNonNull(slaManager);
+ this.stateManager = requireNonNull(stateManager);
+ this.missingMaintenanceCounter = statsProvider.makeCounter(MISSING_MAINTENANCE_REQUEST);
+ }
+
+ private Set<String> drainTasksOnHost(String host, StoreProvider store) {
+ Query.Builder query = Query.slaveScoped(host).active();
+
+ List<IScheduledTask> candidates = new ArrayList<>(store.getTaskStore().fetchTasks(query));
+
+ if (candidates.isEmpty()) {
+ LOG.info("No tasks to drain on host: {}", host);
+ return Collections.emptySet();
+ }
+
+ // shuffle the candidates to avoid head-of-line blocking
+ Collections.shuffle(candidates);
+ candidates.forEach(task -> drainTask(task, store));
+
+ return candidates.stream().map(Tasks::id).collect(Collectors.toSet());
+ }
+
+ private Set<IHostStatus> watchDrainingTasks(MutableStoreProvider store, Set<String> hosts) {
+ LOG.info("Hosts to drain: " + hosts);
+ Set<String> emptyHosts = Sets.newHashSet();
+ for (String host : hosts) {
+ Set<String> drainedTasks = drainTasksOnHost(host, store);
+ // If there are no tasks on the host, immediately transition to DRAINED.
+ if (drainedTasks.isEmpty()) {
+ emptyHosts.add(host);
+ }
+ }
+
+ return ImmutableSet.<IHostStatus>builder()
+ .addAll(setMaintenanceMode(store, emptyHosts, DRAINED))
+ .addAll(setMaintenanceMode(store, Sets.difference(hosts, emptyHosts), DRAINING))
+ .build();
+ }
+
+ /**
+ * Notifies the MaintenanceController that a task has changed state.
+ *
+ * @param change Event
+ */
+ @Subscribe
+ public void taskChangedState(final TaskStateChange change) {
+ if (Tasks.isTerminated(change.getNewState())) {
+ final String host = change.getTask().getAssignedTask().getSlaveHost();
+ batchWorker.execute(store -> {
+ // If the task _was_ associated with a draining host, and it was the last task on the
+ // host.
+ Optional<IHostAttributes> attributes =
+ store.getAttributeStore().getHostAttributes(host);
+ if (attributes.isPresent() && attributes.get().getMode() == DRAINING) {
+ Query.Builder builder = Query.slaveScoped(host).active();
+ Iterable<IScheduledTask> activeTasks = store.getTaskStore().fetchTasks(builder);
+ if (Iterables.isEmpty(activeTasks)) {
+ LOG.info("Moving host {} into DRAINED", host);
+ setMaintenanceMode(store, ImmutableSet.of(host), DRAINED);
+ } else {
+ LOG.info("Host {} is DRAINING with active tasks: {}", host, Tasks.ids(activeTasks));
+ }
+ }
+ return BatchWorker.NO_RESULT;
+ });
+ }
+ }
+
+ @Override
+ public Set<IHostStatus> startMaintenance(Set<String> hosts) {
+ return storage.write(
+ storeProvider -> setMaintenanceMode(storeProvider, hosts, MaintenanceMode.SCHEDULED));
+ }
+
+ private void recordMaintenanceRequests(
+ MutableStoreProvider store,
+ Set<String> hosts,
+ SlaPolicy defaultSlaPolicy,
+ long timeoutSecs) {
+
+ hosts.forEach(
+ host -> store.getHostMaintenanceStore().saveHostMaintenanceRequest(
+ IHostMaintenanceRequest.build(
+ new HostMaintenanceRequest()
+ .setHost(host)
+ .setDefaultSlaPolicy(defaultSlaPolicy)
+ .setTimeoutSecs(timeoutSecs)
+ .setCreatedTimestampMs(System.currentTimeMillis()))));
+ }
+
+ @Override
+ public Set<IHostStatus> drain(Set<String> hosts) {
+ return storage.write(store -> {
+ // Create a dummy maintenance request zero percent sla and timeout to force drain.
+ recordMaintenanceRequests(store, hosts, ZERO_PERCENT_SLA, 0);
+ return watchDrainingTasks(store, hosts);
+ });
+ }
+
+ @Override
+ public Set<IHostStatus> slaDrain(
+ Set<String> hosts,
+ SlaPolicy defaultSlaPolicy,
+ long timeoutSecs) {
+
+ // We can have only one maintenance request per host at any time.
+ // So we will simply overwrite any existing request. If the current one is actively handled,
+ // during the write, the new one will just be a no-op, since the host is already being
+ // drained. If host is in DRAINED it will be moved back into DRAINING and then back into
+ // DRAINED without having to perform any work.
+ return storage.write(store -> {
+ recordMaintenanceRequests(store, hosts, defaultSlaPolicy, timeoutSecs);
+ return setMaintenanceMode(store, hosts, DRAINING);
+ });
+ }
+
+ private Optional<String> getHostname(Protos.InverseOffer offer) {
+ if (offer.getUrl().getAddress().hasHostname()) {
+ return Optional.of(offer.getUrl().getAddress().getHostname());
+ } else {
+ return Optional.empty();
+ }
+ }
+
+ @Override
+ public void drainForInverseOffer(Protos.InverseOffer offer) {
+ // TaskStore does not allow for querying by agent id.
+ Optional<String> hostname = getHostname(offer);
+
+ if (hostname.isPresent()) {
+ String host = hostname.get();
+ storage.write(storeProvider -> {
+ // Create a dummy maintenance request zero percent sla and timeout to force drain.
+ recordMaintenanceRequests(storeProvider, ImmutableSet.of(host), ZERO_PERCENT_SLA, 0);
+ return drainTasksOnHost(host, storeProvider);
+ });
+ } else {
+ LOG.error("Unable to drain tasks on agent {} because "
+ + "no hostname attached to inverse offer {}.", offer.getAgentId(), offer.getId());
+ }
+ }
+
+ private static final Function<IHostAttributes, String> HOST_NAME =
+ IHostAttributes::getHost;
+
+ private static final Function<IHostAttributes, IHostStatus> ATTRS_TO_STATUS =
+ attributes -> IHostStatus.build(
+ new HostStatus().setHost(attributes.getHost()).setMode(attributes.getMode()));
+
+ private static final Function<IHostStatus, MaintenanceMode> GET_MODE = IHostStatus::getMode;
+
+ @Override
+ public MaintenanceMode getMode(final String host) {
+ return storage.read(storeProvider -> storeProvider.getAttributeStore().getHostAttributes(host)
+ .map(ATTRS_TO_STATUS)
+ .map(GET_MODE)
+ .orElse(MaintenanceMode.NONE));
+ }
+
+ @Override
+ public Set<IHostStatus> getStatus(final Set<String> hosts) {
+ return storage.read(storeProvider -> {
+ // Warning - this is filtering _all_ host attributes. If using this to frequently query
+ // for a small set of hosts, a getHostAttributes variant should be added.
+ return storeProvider.getAttributeStore().getHostAttributes().stream()
+ .filter(Predicates.compose(Predicates.in(hosts), HOST_NAME))
+ .map(ATTRS_TO_STATUS)
+ .collect(Collectors.toSet());
+ });
+ }
+
+ @Override
+ public Set<IHostStatus> endMaintenance(final Set<String> hosts) {
+ return storage.write(
+ storeProvider -> {
+ hosts.forEach(
+ h -> storeProvider.getHostMaintenanceStore().removeHostMaintenanceRequest(h));
+ return setMaintenanceMode(storeProvider, hosts, MaintenanceMode.NONE);
+ });
+ }
+
+ private Set<IHostStatus> setMaintenanceMode(
+ MutableStoreProvider storeProvider,
+ Set<String> hosts,
+ MaintenanceMode mode) {
+
+ AttributeStore.Mutable store = storeProvider.getAttributeStore();
+ ImmutableSet.Builder<IHostStatus> statuses = ImmutableSet.builder();
+ for (String host : hosts) {
+ LOG.info("Setting maintenance mode to {} for host {}", mode, host);
+ Optional<IHostAttributes> toSave = AttributeStore.Util.mergeMode(store, host, mode);
+ if (toSave.isPresent()) {
+ store.saveHostAttributes(toSave.get());
+ LOG.info("Updated host attributes: " + toSave.get());
+ statuses.add(IHostStatus.build(new HostStatus().setHost(host).setMode(mode)));
+ }
+ }
+ return statuses.build();
+ }
+
+ @VisibleForTesting
+ void runForTest() {
+ runOneIteration();
+ }
+
+ @Timed
+ @Override
+ protected void runOneIteration() {
+ LOG.info("Looking for hosts in DRAINING state");
+ storage.read(store -> {
+ store.getAttributeStore()
+ .getHostAttributes()
+ .stream()
+ .filter(h -> h.getMode() == DRAINING)
+ .forEach(h -> {
+ if (drainTasksOnHost(h.getHost(), store).isEmpty()) {
+ storage.write(mutable -> setMaintenanceMode(
+ mutable,
+ ImmutableSet.of(h.getHost()),
+ DRAINED));
+ }
+ });
+ return null;
+ });
+ }
+
+ @Override
+ protected Scheduler scheduler() {
+ return Scheduler.newFixedDelaySchedule(
+ pollingInterval.getValue(),
+ pollingInterval.getValue(),
+ pollingInterval.getUnit().getTimeUnit());
+ }
+
+ private void drainTask(IScheduledTask task, StoreProvider store) {
+ String host = task.getAssignedTask().getSlaveHost();
+ Optional<IHostMaintenanceRequest> hostMaintenanceRequest =
+ store.getHostMaintenanceStore().getHostMaintenanceRequest(host);
+ if (!hostMaintenanceRequest.isPresent()) {
+ LOG.error("No maintenance request found for host: {}. Assuming SLA not satisfied.", host);
+ missingMaintenanceCounter.incrementAndGet();
+ return;
+ }
+
+ boolean force = false;
+ long expireMs =
+ System.currentTimeMillis() - hostMaintenanceRequest.get().getCreatedTimestampMs();
+ if (hostMaintenanceRequest.get().getTimeoutSecs()
+ < TimeAmount.of(expireMs, Time.MILLISECONDS).as(Time.SECONDS)) {
+ LOG.warn("Maintenance request timed out for host: {} after {} secs. Forcing drain of {}.",
+ host, hostMaintenanceRequest.get().getTimeoutSecs(), Tasks.id(task));
+ force = true;
+ }
+
+ final ISlaPolicy slaPolicy = task.getAssignedTask().getTask().isSetSlaPolicy()
+ ? task.getAssignedTask().getTask().getSlaPolicy()
+ : hostMaintenanceRequest.get().getDefaultSlaPolicy();
+
+ slaManager.checkSlaThenAct(
+ task,
+ slaPolicy,
+ storeProvider -> stateManager.changeState(
+ storeProvider,
+ Tasks.id(task),
+ Optional.empty(),
+ ScheduleStatus.DRAINING,
+ Optional.of(DRAINING_MESSAGE)),
+ force);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/java/org/apache/aurora/scheduler/maintenance/MaintenanceModule.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/maintenance/MaintenanceModule.java b/src/main/java/org/apache/aurora/scheduler/maintenance/MaintenanceModule.java
new file mode 100644
index 0000000..f4f4ec3
--- /dev/null
+++ b/src/main/java/org/apache/aurora/scheduler/maintenance/MaintenanceModule.java
@@ -0,0 +1,78 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.maintenance;
+
+import java.util.concurrent.ScheduledExecutorService;
+import javax.inject.Singleton;
+
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.Parameters;
+import com.google.inject.AbstractModule;
+import com.google.inject.PrivateModule;
+import com.google.inject.TypeLiteral;
+
+import org.apache.aurora.common.quantity.Amount;
+import org.apache.aurora.common.quantity.Time;
+import org.apache.aurora.scheduler.SchedulerServicesModule;
+import org.apache.aurora.scheduler.base.AsyncUtil;
+import org.apache.aurora.scheduler.config.types.TimeAmount;
+import org.apache.aurora.scheduler.config.validators.PositiveAmount;
+import org.apache.aurora.scheduler.events.PubsubEventModule;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Binding module for maintenance related logic.
+ */
+public class MaintenanceModule extends AbstractModule {
+
+ private static final Logger LOG = LoggerFactory.getLogger(MaintenanceModule.class);
+
+ @Parameters(separators = "=")
+ public static class Options {
+ @Parameter(names = "-host_maintenance_polling_interval",
+ validateValueWith = PositiveAmount.class,
+ description = "Interval between polling for pending host maintenance requests.")
+ public TimeAmount hostMaintenancePollingInterval = new TimeAmount(1, Time.MINUTES);
+ }
+
+ private final Options options;
+
+ public MaintenanceModule(final Options options) {
+ this.options = options;
+ }
+
+ @Override
+ protected void configure() {
+ bind(new TypeLiteral<Amount<Long, Time>>() { })
+ .annotatedWith(
+ MaintenanceController.MaintenanceControllerImpl.PollingInterval.class)
+ .toInstance(options.hostMaintenancePollingInterval);
+ bind(MaintenanceController.class).to(MaintenanceController.MaintenanceControllerImpl.class);
+ bind(MaintenanceController.MaintenanceControllerImpl.class).in(Singleton.class);
+ PubsubEventModule.bindSubscriber(
+ binder(),
+ MaintenanceController.MaintenanceControllerImpl.class);
+
+ install(new PrivateModule() {
+ @Override
+ protected void configure() {
+ bind(ScheduledExecutorService.class).toInstance(
+ AsyncUtil.singleThreadLoggingScheduledExecutor("MaintenanceController-%d", LOG));
+ }
+ });
+ SchedulerServicesModule.addSchedulerActiveServiceBinding(binder())
+ .to(MaintenanceController.MaintenanceControllerImpl.class);
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/java/org/apache/aurora/scheduler/mesos/MesosCallbackHandler.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/mesos/MesosCallbackHandler.java b/src/main/java/org/apache/aurora/scheduler/mesos/MesosCallbackHandler.java
index 3b4df55..53b14e9 100644
--- a/src/main/java/org/apache/aurora/scheduler/mesos/MesosCallbackHandler.java
+++ b/src/main/java/org/apache/aurora/scheduler/mesos/MesosCallbackHandler.java
@@ -39,10 +39,10 @@ import org.apache.aurora.scheduler.base.SchedulerException;
import org.apache.aurora.scheduler.events.EventSink;
import org.apache.aurora.scheduler.events.PubsubEvent;
import org.apache.aurora.scheduler.events.PubsubEventModule;
+import org.apache.aurora.scheduler.maintenance.MaintenanceController;
import org.apache.aurora.scheduler.offers.HostOffer;
import org.apache.aurora.scheduler.offers.OfferManager;
import org.apache.aurora.scheduler.offers.OfferManagerModule;
-import org.apache.aurora.scheduler.state.MaintenanceController;
import org.apache.aurora.scheduler.storage.AttributeStore;
import org.apache.aurora.scheduler.storage.Storage;
import org.apache.aurora.scheduler.storage.entities.IHostAttributes;
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/java/org/apache/aurora/scheduler/sla/SlaManager.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/sla/SlaManager.java b/src/main/java/org/apache/aurora/scheduler/sla/SlaManager.java
new file mode 100644
index 0000000..98bec48
--- /dev/null
+++ b/src/main/java/org/apache/aurora/scheduler/sla/SlaManager.java
@@ -0,0 +1,437 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.sla;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
+import java.util.stream.Collectors;
+import javax.inject.Qualifier;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.Striped;
+import com.google.gson.Gson;
+import com.google.inject.Inject;
+
+import org.apache.aurora.common.inject.TimedInterceptor.Timed;
+import org.apache.aurora.common.quantity.Amount;
+import org.apache.aurora.common.quantity.Time;
+import org.apache.aurora.common.stats.StatsProvider;
+import org.apache.aurora.gen.ScheduleStatus;
+import org.apache.aurora.scheduler.TierManager;
+import org.apache.aurora.scheduler.base.Query;
+import org.apache.aurora.scheduler.base.Tasks;
+import org.apache.aurora.scheduler.config.types.TimeAmount;
+import org.apache.aurora.scheduler.storage.Storage;
+import org.apache.aurora.scheduler.storage.Storage.StoreProvider;
+import org.apache.aurora.scheduler.storage.entities.ICoordinatorSlaPolicy;
+import org.apache.aurora.scheduler.storage.entities.IJobKey;
+import org.apache.aurora.scheduler.storage.entities.IScheduledTask;
+import org.apache.aurora.scheduler.storage.entities.IServerInfo;
+import org.apache.aurora.scheduler.storage.entities.ISlaPolicy;
+import org.apache.aurora.scheduler.storage.entities.ITaskEvent;
+import org.apache.thrift.TException;
+import org.apache.thrift.TSerializer;
+import org.apache.thrift.protocol.TSimpleJSONProtocol;
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.Param;
+import org.asynchttpclient.Response;
+import org.asynchttpclient.util.HttpConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static java.lang.annotation.ElementType.FIELD;
+import static java.lang.annotation.ElementType.METHOD;
+import static java.lang.annotation.ElementType.PARAMETER;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Provides methods for performing SLA-safe work. It is used for maintenance and job update
+ * operations to guarantee that a job's SLA requirements are always satisfied.
+ */
+public class SlaManager {
+ private static final Logger LOG = LoggerFactory.getLogger(SlaManager.class);
+ @VisibleForTesting
+ @Qualifier
+ @Target({ FIELD, PARAMETER, METHOD }) @Retention(RUNTIME)
+ @interface SlaManagerExecutor { }
+
+ @VisibleForTesting
+ @Qualifier
+ @Target({ FIELD, PARAMETER, METHOD }) @Retention(RUNTIME)
+ @interface MaxParallelCoordinators { }
+
+ @VisibleForTesting
+ @Qualifier
+ @Target({ FIELD, PARAMETER, METHOD }) @Retention(RUNTIME)
+ @interface HttpClient { }
+
+ @VisibleForTesting
+ @Qualifier
+ @Target({ FIELD, PARAMETER, METHOD }) @Retention(RUNTIME)
+ @interface MinRequiredInstances { }
+
+ @VisibleForTesting
+ static final String TASK_PARAM = "task";
+
+ private static final String ATTEMPTS_STAT_NAME = "sla_coordinator_attempts";
+ private static final String SUCCESS_STAT_NAME = "sla_coordinator_success";
+ private static final String ERRORS_STAT_NAME = "sla_coordinator_errors";
+ private static final String USER_ERRORS_STAT_NAME = "sla_coordinator_user_errors";
+ private static final String LOCK_STARVATION_STAT_NAME = "sla_coordinator_lock_starvation";
+
+ private final ScheduledExecutorService executor;
+ private final Storage storage;
+ private final IServerInfo serverInfo;
+ private final AsyncHttpClient httpClient;
+ private final Striped<Lock> lock;
+ private final int minRequiredInstances;
+ private final TierManager tierManager;
+
+ private final AtomicLong attemptsCounter;
+ private final AtomicLong successCounter;
+ private final AtomicLong errorsCounter;
+ private final AtomicLong userErrorsCounter;
+ private final AtomicLong lockStarvationCounter;
+ private final LoadingCache<String, AtomicLong> errorsByTaskCounter;
+ private final LoadingCache<String, AtomicLong> userErrorsByTaskCounter;
+ private final LoadingCache<String, AtomicLong> lockStarvationByTaskCounter;
+
+ @Inject
+ SlaManager(@SlaManagerExecutor ScheduledExecutorService executor,
+ @MaxParallelCoordinators Integer maxCoordinatorLocks,
+ @MinRequiredInstances Integer minRequiredInstances,
+ Storage storage,
+ IServerInfo serverInfo,
+ @HttpClient AsyncHttpClient httpClient,
+ TierManager tierManager,
+ StatsProvider statsProvider) {
+
+ this.executor = requireNonNull(executor);
+ this.storage = requireNonNull(storage);
+ this.serverInfo = requireNonNull(serverInfo);
+ this.httpClient = requireNonNull(httpClient);
+ this.tierManager = requireNonNull(tierManager);
+ this.minRequiredInstances = requireNonNull(minRequiredInstances);
+ this.attemptsCounter = statsProvider.makeCounter(ATTEMPTS_STAT_NAME);
+ this.successCounter = statsProvider.makeCounter(SUCCESS_STAT_NAME);
+ this.errorsCounter = statsProvider.makeCounter(ERRORS_STAT_NAME);
+ this.userErrorsCounter = statsProvider.makeCounter(USER_ERRORS_STAT_NAME);
+ this.lockStarvationCounter = statsProvider.makeCounter(LOCK_STARVATION_STAT_NAME);
+ this.lock = Striped.lazyWeakLock(requireNonNull(maxCoordinatorLocks));
+ this.errorsByTaskCounter = CacheBuilder.newBuilder().build(
+ new CacheLoader<String, AtomicLong>() {
+ @Override
+ public AtomicLong load(String key) {
+ return statsProvider.makeCounter(key);
+ }
+ }
+ );
+ this.userErrorsByTaskCounter = CacheBuilder.newBuilder().build(
+ new CacheLoader<String, AtomicLong>() {
+ @Override
+ public AtomicLong load(String key) {
+ return statsProvider.makeCounter(key);
+ }
+ }
+ );
+ this.lockStarvationByTaskCounter = CacheBuilder.newBuilder().build(
+ new CacheLoader<String, AtomicLong>() {
+ @Override
+ public AtomicLong load(String key) {
+ return statsProvider.makeCounter(key);
+ }
+ }
+ );
+ }
+
+ private long getSlaDuration(ISlaPolicy slaPolicy) {
+ if (slaPolicy.isSetPercentageSlaPolicy()) {
+ return slaPolicy.getPercentageSlaPolicy().getDurationSecs();
+ } else if (slaPolicy.isSetCountSlaPolicy()) {
+ return slaPolicy.getCountSlaPolicy().getDurationSecs();
+ }
+
+ throw new IllegalArgumentException("Expected a percentage/count sla policy.");
+ }
+
+ private boolean meetsSLAInstances(ISlaPolicy slaPolicy, long running, long total) {
+ if (slaPolicy.isSetPercentageSlaPolicy()) {
+ double percentageRunning = ((double) running / total) * 100.0;
+ double percentageRequired = slaPolicy.getPercentageSlaPolicy().getPercentage();
+ long numMaintenance = (long) ((1 - percentageRequired / 100.0) * total);
+ if (numMaintenance < 1) {
+ double updated = ((total - 1) / (double) total) * 100;
+ LOG.warn("Invalid PercentageSlaPolicy(percentage={}) for task with {} instances "
+ + "that allows {} instances to be in maintenance."
+ + "Using percentage={} which allows 1 instance to be under maintenance.",
+ percentageRequired, total, numMaintenance, updated);
+ percentageRequired = updated;
+ }
+ return percentageRunning >= percentageRequired;
+ } else if (slaPolicy.isSetCountSlaPolicy()) {
+ long numRequired = slaPolicy.getCountSlaPolicy().getCount();
+ long numMaintenance = total - numRequired;
+ if (numMaintenance < 1) {
+ long updated = total - 1;
+ LOG.warn("Invalid CountSlaPolicy(count={}) for task with {} instances "
+ + "that allows {} instances to be in maintenance."
+ + "Using count={} which allows 1 instance to be under maintenance.",
+ numRequired, total, numMaintenance, updated);
+ numRequired = updated;
+ }
+ return running >= numRequired;
+ }
+
+ throw new IllegalArgumentException("Expected a percentage/count sla policy.");
+ }
+
+ private boolean meetsSLADuration(IScheduledTask task, Amount<Long, Time> slaDuration) {
+ return task.getTaskEvents()
+ .stream()
+ .filter(te -> te.getStatus() == ScheduleStatus.RUNNING)
+ .map(ITaskEvent::getTimestamp)
+ .max(Long::compare)
+ .map(t -> System.currentTimeMillis() - t > slaDuration.as(Time.MILLISECONDS))
+ .orElse(false);
+ }
+
+ private boolean checkSla(IScheduledTask task, ISlaPolicy slaPolicy, StoreProvider store) {
+ // Find the number of active tasks for the job, it will be used as the set of tasks
+ // against which the percentage/count based SLA will be calculated against.
+ final long numActive = store.getTaskStore().fetchTasks(
+ Query.jobScoped(task.getAssignedTask().getTask().getJob()).active()
+ ).size();
+
+ if (skipSla(task, numActive)) {
+ LOG.info("Skip SLA for {} because it is not production or does not have enough instances.",
+ Tasks.id(task));
+ return true;
+ }
+
+ // Find tasks which have been RUNNING for the required SLA duration.
+ Amount<Long, Time> slaDuration = new TimeAmount(getSlaDuration(slaPolicy), Time.SECONDS);
+ final Set<IScheduledTask> running = store.getTaskStore().fetchTasks(
+ Query.jobScoped(task.getAssignedTask().getTask().getJob())
+ .byStatus(ScheduleStatus.RUNNING))
+ .stream()
+ .filter(t -> !Tasks.id(t).equals(Tasks.id(task))) // exclude the task to be removed
+ .filter(t -> meetsSLADuration(t, slaDuration)) // task is running for sla duration
+ .collect(Collectors.toSet());
+
+ // Check it we satisfy the number of RUNNING tasks per duration time.
+ boolean meetsSla = meetsSLAInstances(slaPolicy, running.size(), numActive);
+
+ LOG.info("SlaCheck: {}, {} tasks unaffected after updating state for {}.",
+ meetsSla,
+ running.size(),
+ Tasks.id(task));
+
+ return meetsSla;
+ }
+
+ /**
+ * Performs the supplied {@link Storage.MutateWork} after checking with the configured
+ * coordinator endpoint to make sure it is safe to perform the work.
+ *
+ * NOTE: Both the SLA check and the {@link Storage.MutateWork} will be performed within a
+ * {@link Lock} that is indexed by the coordinator url. We do this to make sure that mutations
+ * to the SLA are performed atomically, so the Coordinator does not have to track
+ * concurrent requests and simulate SLA changes.
+ *
+ * @param task Task whose SLA is to checked.
+ * @param slaPolicy {@link ICoordinatorSlaPolicy} to use for checking SLA.
+ * @param work {@link Storage.MutateWork} to perform, if SLA is satisfied.
+ * @param <T> The type of result the {@link Storage.MutateWork} produces.
+ * @param <E> The type of exception the {@link Storage.MutateWork} throw.
+ */
+ private <T, E extends Exception> void askCoordinatorThenAct(
+ IScheduledTask task,
+ ICoordinatorSlaPolicy slaPolicy,
+ Storage.MutateWork<T, E> work) {
+
+ String taskKey = getTaskKey(task);
+
+ LOG.debug("Awaiting lock on coordinator: {} for task: {}",
+ slaPolicy.getCoordinatorUrl(),
+ taskKey);
+ Lock l = lock.get(slaPolicy.getCoordinatorUrl());
+ if (l.tryLock()) {
+ try {
+ LOG.info("Acquired lock on coordinator: {} for task: {}",
+ slaPolicy.getCoordinatorUrl(),
+ taskKey);
+ attemptsCounter.incrementAndGet();
+
+ if (coordinatorAllows(task, taskKey, slaPolicy)) {
+ LOG.info("Performing work after coordinator: {} approval for task: {}",
+ slaPolicy.getCoordinatorUrl(),
+ taskKey);
+ storage.write(work);
+ }
+ } catch (RuntimeException e) {
+ LOG.error("Unexpected failure during coordinator sla check against: {} for task: {}",
+ slaPolicy.getCoordinatorUrl(),
+ taskKey,
+ e);
+ throw e;
+ } catch (Exception e) {
+ incrementErrorCount(ERRORS_STAT_NAME, taskKey);
+ LOG.error("Failed to talk to coordinator: {} for task: {}",
+ slaPolicy.getCoordinatorUrl(),
+ taskKey,
+ e);
+ } finally {
+ LOG.info("Releasing lock for coordinator: {} and task: {}",
+ slaPolicy.getCoordinatorUrl(),
+ taskKey);
+ l.unlock();
+ }
+ } else {
+ incrementErrorCount(LOCK_STARVATION_STAT_NAME, taskKey);
+ LOG.info("Failed to acquire lock on coordinator: {} for task: {}",
+ slaPolicy.getCoordinatorUrl(),
+ taskKey);
+ }
+ }
+
+ private boolean coordinatorAllows(
+ IScheduledTask task,
+ String taskKey,
+ ICoordinatorSlaPolicy slaPolicy)
+ throws InterruptedException, ExecutionException, TException {
+
+ LOG.info("Checking coordinator: {} for task: {}", slaPolicy.getCoordinatorUrl(), taskKey);
+
+ Response response = httpClient.preparePost(slaPolicy.getCoordinatorUrl())
+ .setQueryParams(ImmutableList.of(new Param(TASK_PARAM, taskKey)))
+ .setBody(new TSerializer(new TSimpleJSONProtocol.Factory()).toString(task.newBuilder()))
+ .execute()
+ .get();
+
+ if (response.getStatusCode() != HttpConstants.ResponseStatusCodes.OK_200) {
+ LOG.error("Request failed to coordinator: {} for task: {}. Response: {}",
+ slaPolicy.getCoordinatorUrl(),
+ taskKey,
+ response.getStatusCode());
+ incrementErrorCount(USER_ERRORS_STAT_NAME, taskKey);
+ return false;
+ }
+
+ successCounter.incrementAndGet();
+ String json = response.getResponseBody();
+ LOG.info("Got response: {} from {} for task: {}",
+ json,
+ slaPolicy.getCoordinatorUrl(),
+ taskKey);
+
+ Map<String, Boolean> result = new Gson().fromJson(
+ json,
+ new TypeReference<Map<String, Boolean>>() { }.getType());
+
+ return result.get(slaPolicy.isSetStatusKey() ? slaPolicy.getStatusKey() : "drain");
+ }
+
+ @VisibleForTesting
+ String getTaskKey(IScheduledTask task) {
+ IJobKey jobKey = Tasks.getJob(task);
+ int instanceId = Tasks.getInstanceId(task);
+ return String.format("%s/%s/%s/%s/%s", serverInfo.getClusterName(),
+ jobKey.getRole(), jobKey.getEnvironment(), jobKey.getName(), instanceId);
+ }
+
+ private void incrementErrorCount(String prefix, String taskKey) {
+ try {
+ if (prefix.equals(ERRORS_STAT_NAME)) {
+ errorsCounter.incrementAndGet();
+ errorsByTaskCounter.get(prefix + "_" + taskKey).incrementAndGet();
+ }
+
+ if (prefix.equals(USER_ERRORS_STAT_NAME)) {
+ userErrorsCounter.incrementAndGet();
+ userErrorsByTaskCounter.get(prefix + "_" + taskKey).incrementAndGet();
+ }
+
+ if (prefix.equals(LOCK_STARVATION_STAT_NAME)) {
+ lockStarvationCounter.incrementAndGet();
+ lockStarvationByTaskCounter.get(prefix + "_" + taskKey).incrementAndGet();
+ }
+ } catch (ExecutionException e) {
+ LOG.error("Failed increment failure metrics for task: {}", taskKey, e);
+ }
+ }
+
+ /**
+ * Checks the SLA for the given task using the {@link ISlaPolicy} supplied.
+ *
+ * @param task Task whose SLA is to be checked.
+ * @param slaPolicy {@link ISlaPolicy} to use.
+ * @param work {@link Storage.MutateWork} to perform, if SLA is satisfied.
+ * @param force boolean to indicate if work should be performed without checking SLA.
+ * @param <T> The type of result the {@link Storage.MutateWork} produces.
+ * @param <E> The type of exception the {@link Storage.MutateWork} throw.
+ * @throws E raises exception of type E if the {@link Storage.MutateWork} throws.
+ */
+ @Timed
+ public <T, E extends Exception> void checkSlaThenAct(
+ IScheduledTask task,
+ ISlaPolicy slaPolicy,
+ Storage.MutateWork<T, E> work,
+ boolean force) throws E {
+
+ if (force) {
+ LOG.info("Forcing work without applying SlaPolicy: {} for {}", slaPolicy, Tasks.id(task));
+ storage.write(work);
+ return;
+ }
+
+ LOG.info("Using SlaPolicy: {} for {}", slaPolicy, Tasks.id(task));
+
+ // has custom coordinator sla policy
+ if (slaPolicy.isSetCoordinatorSlaPolicy()) {
+ // schedule work to perform coordinated transition
+ executor.execute(() -> askCoordinatorThenAct(
+ task,
+ slaPolicy.getCoordinatorSlaPolicy(),
+ work));
+ } else {
+ // verify sla and perform work if satisfied
+ storage.write(store -> {
+ if (checkSla(task, slaPolicy, store)) {
+ work.apply(store);
+ }
+ return null; // TODO(sshanmugham) we need to satisfy the interface, refactor later
+ });
+ }
+ }
+
+ private boolean skipSla(IScheduledTask task, long numActive) {
+ if (!tierManager.getTier(task.getAssignedTask().getTask()).isPreemptible()
+ && !tierManager.getTier(task.getAssignedTask().getTask()).isRevocable()) {
+ return numActive < minRequiredInstances;
+ }
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/java/org/apache/aurora/scheduler/sla/SlaModule.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/sla/SlaModule.java b/src/main/java/org/apache/aurora/scheduler/sla/SlaModule.java
index 25ed474..07082a9 100644
--- a/src/main/java/org/apache/aurora/scheduler/sla/SlaModule.java
+++ b/src/main/java/org/apache/aurora/scheduler/sla/SlaModule.java
@@ -18,7 +18,6 @@ import java.lang.annotation.Target;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-
import javax.inject.Inject;
import javax.inject.Qualifier;
@@ -30,6 +29,7 @@ import com.google.common.collect.ImmutableSet;
import com.google.common.util.concurrent.AbstractIdleService;
import com.google.inject.AbstractModule;
import com.google.inject.Singleton;
+import com.google.inject.TypeLiteral;
import org.apache.aurora.common.quantity.Time;
import org.apache.aurora.scheduler.SchedulerServicesModule;
@@ -39,6 +39,9 @@ import org.apache.aurora.scheduler.config.types.TimeAmount;
import org.apache.aurora.scheduler.config.validators.PositiveAmount;
import org.apache.aurora.scheduler.sla.MetricCalculator.MetricCalculatorSettings;
import org.apache.aurora.scheduler.sla.MetricCalculator.MetricCategory;
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.DefaultAsyncHttpClientConfig;
+import org.asynchttpclient.channel.DefaultKeepAliveStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,6 +54,7 @@ import static java.util.Objects.requireNonNull;
import static org.apache.aurora.scheduler.sla.MetricCalculator.MetricCategory.JOB_UPTIMES;
import static org.apache.aurora.scheduler.sla.MetricCalculator.MetricCategory.MEDIANS;
import static org.apache.aurora.scheduler.sla.MetricCalculator.MetricCategory.PLATFORM_UPTIME;
+import static org.asynchttpclient.Dsl.asyncHttpClient;
/**
* Binding module for the sla processor.
@@ -76,6 +80,27 @@ public class SlaModule extends AbstractModule {
description = "Metric categories collected for non production tasks.",
splitter = CommaSplitter.class)
public List<MetricCategory> slaNonProdMetrics = ImmutableList.of();
+
+ @Parameter(names = "-sla_coordinator_timeout",
+ validateValueWith = PositiveAmount.class,
+ description = "Timeout interval for communicating with Coordinator.")
+ public TimeAmount slaCoordinatorTimeout = new TimeAmount(1, Time.MINUTES);
+
+ @Parameter(names = "-max_parallel_coordinated_maintenance",
+ description = "Maximum number of coordinators that can be contacted in parallel.")
+ public Integer maxParallelCoordinators = 10;
+
+ @Parameter(names = "-min_required_instances_for_sla_check",
+ description = "Minimum number of instances required for a job to be eligible for SLA "
+ + "check. This does not apply to jobs that have a CoordinatorSlaPolicy.")
+ public Integer minRequiredInstances = 20;
+
+ @Parameter(names = "-max_sla_duration_secs",
+ validateValueWith = PositiveAmount.class,
+ description = "Maximum duration window for which SLA requirements are to be satisfied."
+ + "This does not apply to jobs that have a CoordinatorSlaPolicy."
+ )
+ public TimeAmount maxSlaDuration = new TimeAmount(2, Time.HOURS);
}
@VisibleForTesting
@@ -104,6 +129,36 @@ public class SlaModule extends AbstractModule {
bind(SlaUpdater.class).in(Singleton.class);
SchedulerServicesModule.addSchedulerActiveServiceBinding(binder()).to(SlaUpdater.class);
+
+ DefaultAsyncHttpClientConfig config = new DefaultAsyncHttpClientConfig.Builder()
+ .setConnectTimeout(options.slaCoordinatorTimeout.as(Time.MILLISECONDS).intValue())
+ .setHandshakeTimeout(options.slaCoordinatorTimeout.as(Time.MILLISECONDS).intValue())
+ .setSslSessionTimeout(options.slaCoordinatorTimeout.as(Time.MILLISECONDS).intValue())
+ .setReadTimeout(options.slaCoordinatorTimeout.as(Time.MILLISECONDS).intValue())
+ .setRequestTimeout(options.slaCoordinatorTimeout.as(Time.MILLISECONDS).intValue())
+ .setKeepAliveStrategy(new DefaultKeepAliveStrategy())
+ .build();
+ AsyncHttpClient httpClient = asyncHttpClient(config);
+
+ bind(AsyncHttpClient.class)
+ .annotatedWith(SlaManager.HttpClient.class)
+ .toInstance(httpClient);
+
+ bind(new TypeLiteral<Integer>() { })
+ .annotatedWith(SlaManager.MinRequiredInstances.class)
+ .toInstance(options.minRequiredInstances);
+
+ bind(new TypeLiteral<Integer>() { })
+ .annotatedWith(SlaManager.MaxParallelCoordinators.class)
+ .toInstance(options.maxParallelCoordinators);
+
+ bind(ScheduledExecutorService.class)
+ .annotatedWith(SlaManager.SlaManagerExecutor.class)
+ .toInstance(AsyncUtil.loggingScheduledExecutor(
+ options.maxParallelCoordinators,
+ "SlaManager-%d", LOG));
+
+ bind(SlaManager.class).in(javax.inject.Singleton.class);
}
// TODO(ksweeney): This should use AbstractScheduledService.
[2/4] aurora git commit: Enable `Tasks` to specify their own custom
maintenance SLA.
Posted by sa...@apache.org.
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/sla/SlaManagerTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/sla/SlaManagerTest.java b/src/test/java/org/apache/aurora/scheduler/sla/SlaManagerTest.java
new file mode 100644
index 0000000..759a1bc
--- /dev/null
+++ b/src/test/java/org/apache/aurora/scheduler/sla/SlaManagerTest.java
@@ -0,0 +1,1239 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.sla;
+
+import java.io.IOException;
+import java.net.URLEncoder;
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableSet;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.TypeLiteral;
+
+import org.apache.aurora.common.stats.StatsProvider;
+import org.apache.aurora.common.testing.easymock.EasyMockTest;
+import org.apache.aurora.gen.CoordinatorSlaPolicy;
+import org.apache.aurora.gen.CountSlaPolicy;
+import org.apache.aurora.gen.PercentageSlaPolicy;
+import org.apache.aurora.gen.ScheduleStatus;
+import org.apache.aurora.gen.ScheduledTask;
+import org.apache.aurora.gen.ServerInfo;
+import org.apache.aurora.gen.SlaPolicy;
+import org.apache.aurora.scheduler.TierManager;
+import org.apache.aurora.scheduler.base.AsyncUtil;
+import org.apache.aurora.scheduler.base.Query;
+import org.apache.aurora.scheduler.base.TaskTestUtil;
+import org.apache.aurora.scheduler.base.Tasks;
+import org.apache.aurora.scheduler.state.StateManager;
+import org.apache.aurora.scheduler.storage.Storage;
+import org.apache.aurora.scheduler.storage.entities.IScheduledTask;
+import org.apache.aurora.scheduler.storage.entities.IServerInfo;
+import org.apache.aurora.scheduler.storage.entities.ISlaPolicy;
+import org.apache.aurora.scheduler.storage.testing.StorageTestUtil;
+import org.apache.aurora.scheduler.testing.FakeStatsProvider;
+import org.apache.thrift.TException;
+import org.apache.thrift.TSerializer;
+import org.apache.thrift.protocol.TSimpleJSONProtocol;
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.DefaultAsyncHttpClient;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.handler.AbstractHandler;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.aurora.gen.ScheduleStatus.PENDING;
+import static org.apache.aurora.gen.ScheduleStatus.RUNNING;
+import static org.apache.aurora.scheduler.base.TaskTestUtil.TIER_MANAGER;
+import static org.easymock.EasyMock.expect;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.fail;
+
+public class SlaManagerTest extends EasyMockTest {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SlaManagerTest.class);
+ private static final String CLUSTER_NAME = "test_cluster";
+ private static final String STATS_URL_PREFIX = "fake_url";
+ private static final String HOST_A = "a";
+
+ private static final ISlaPolicy PERCENTAGE_SLA_POLICY = ISlaPolicy.build(
+ SlaPolicy.percentageSlaPolicy(
+ new PercentageSlaPolicy()
+ .setPercentage(66)
+ .setDurationSecs(1800)));
+
+ private static final ISlaPolicy COUNT_SLA_POLICY = ISlaPolicy.build(
+ SlaPolicy.countSlaPolicy(
+ new CountSlaPolicy()
+ .setCount(2)
+ .setDurationSecs(1800)));
+
+ private AsyncHttpClient httpClient;
+ private SlaManager slaManager;
+ private StorageTestUtil storageUtil;
+ private StateManager stateManager;
+ private IServerInfo serverInfo;
+ private Server jettyServer;
+ private CountDownLatch coordinatorResponded;
+
+ @Before
+ public void setUp() {
+ jettyServer = new Server(0); // Start Jetty server with ephemeral port
+ storageUtil = new StorageTestUtil(this);
+ storageUtil.expectOperations();
+ stateManager = createMock(StateManager.class);
+ httpClient = new DefaultAsyncHttpClient();
+ coordinatorResponded = new CountDownLatch(1);
+
+ serverInfo = IServerInfo.build(
+ new ServerInfo()
+ .setClusterName(CLUSTER_NAME)
+ .setStatsUrlPrefix(STATS_URL_PREFIX));
+
+ Injector injector = Guice.createInjector(
+ new AbstractModule() {
+ @Override
+ protected void configure() {
+ bind(Storage.class).toInstance(storageUtil.storage);
+ bind(StateManager.class).toInstance(stateManager);
+ bind(StatsProvider.class).toInstance(new FakeStatsProvider());
+ bind(TierManager.class).toInstance(TIER_MANAGER);
+ bind(AsyncHttpClient.class)
+ .annotatedWith(SlaManager.HttpClient.class)
+ .toInstance(httpClient);
+
+ bind(new TypeLiteral<Integer>() { })
+ .annotatedWith(SlaManager.MinRequiredInstances.class)
+ .toInstance(2);
+
+ bind(new TypeLiteral<Integer>() { })
+ .annotatedWith(SlaManager.MaxParallelCoordinators.class)
+ .toInstance(10);
+
+ bind(ScheduledExecutorService.class)
+ .annotatedWith(SlaManager.SlaManagerExecutor.class)
+ .toInstance(AsyncUtil.loggingScheduledExecutor(
+ 10, "SlaManagerTest-%d", LOG));
+
+ bind(IServerInfo.class).toInstance(serverInfo);
+ }
+ }
+ );
+ slaManager = injector.getInstance(SlaManager.class);
+
+ addTearDown(() -> jettyServer.stop());
+ }
+
+ private static IScheduledTask makeTask(
+ String taskId,
+ int instanceId,
+ ScheduleStatus status) {
+ return makeTask(taskId, instanceId, status, 1000, true);
+ }
+
+ private static IScheduledTask makeTask(
+ String taskId,
+ int instanceId,
+ long runningSince) {
+ return makeTask(taskId, instanceId, RUNNING, runningSince, true);
+ }
+
+ private static IScheduledTask makeTask(
+ String taskId,
+ int instanceId,
+ ScheduleStatus status,
+ long runningSince,
+ boolean prod) {
+ ScheduledTask builder = TaskTestUtil.addStateTransition(
+ TaskTestUtil.makeTask(taskId, TaskTestUtil.JOB, instanceId, prod),
+ status,
+ runningSince).newBuilder();
+ builder.getAssignedTask().setSlaveHost(SlaManagerTest.HOST_A);
+ return IScheduledTask.build(builder);
+ }
+
+ /**
+ * Verifies that SLA check passes and the supplied {@link Storage.MutateWork} gets executed
+ * for a job that has {@link CountSlaPolicy#count} + 1 tasks that have been RUNNING for
+ * the required {@link CountSlaPolicy#durationSecs} with {@link CountSlaPolicy#count} set to 2.
+ */
+ @Test
+ public void testCheckCountSlaPassesThenActs() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ IScheduledTask task2 = makeTask("taskB", 2, RUNNING);
+ IScheduledTask task3 = makeTask("taskC", 3, RUNNING);
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // mock calls to fetch all RUNNING tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).byStatus(RUNNING)))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // expect that the fetchTask in the work is called, after sla check passes
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ COUNT_SLA_POLICY,
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that SLA check gets overridden, passes and the supplied {@link Storage.MutateWork} is
+ * executed for a job that has and aggressive {@link CountSlaPolicy#count} > total number of
+ * instances.
+ */
+ @Test
+ public void testCheckCountSlaOverridesAggressiveCountPassesThenActs() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ IScheduledTask task2 = makeTask("taskB", 2, RUNNING);
+ IScheduledTask task3 = makeTask("taskC", 3, RUNNING);
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // mock calls to fetch all RUNNING tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).byStatus(RUNNING)))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // expect that the fetchTask in the work is called, after sla check passes
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ ISlaPolicy.build(
+ SlaPolicy.countSlaPolicy(
+ new CountSlaPolicy()
+ .setCount(3)
+ .setDurationSecs(1800))),
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that SLA check gets overridden, fails and the supplied {@link Storage.MutateWork} is
+ * not executed for a job that has and aggressive {@link CountSlaPolicy#count} > total number of
+ * instances.
+ */
+ @Test
+ public void testCheckCountSlaOverridesAggressiveCountFailsAct() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ IScheduledTask task2 = makeTask("taskB", 2, RUNNING);
+ IScheduledTask task3 = makeTask("taskC", 3, PENDING);
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // mock calls to fetch all RUNNING tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).byStatus(RUNNING)))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ ISlaPolicy.build(
+ SlaPolicy.countSlaPolicy(
+ new CountSlaPolicy()
+ .setCount(3)
+ .setDurationSecs(1800))),
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that SLA check fails and the supplied {@link Storage.MutateWork} does not get
+ * executed for a job that has {@link CountSlaPolicy#count} tasks that have been RUNNING for
+ * the required {@link CountSlaPolicy#durationSecs} and 1 task in PENDING with
+ * {@link CountSlaPolicy#count} set to 2.
+ */
+ @Test
+ public void testCheckCountSlaFailsDoesNotAct() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ IScheduledTask task2 = makeTask("taskB", 2, RUNNING);
+ IScheduledTask task3 = makeTask("taskC", 3, PENDING);
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // mock calls to fetch all RUNNING tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).byStatus(RUNNING)))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ COUNT_SLA_POLICY,
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that SLA check is skipped and the supplied {@link Storage.MutateWork} gets
+ * executed for a job that does not meet {@link SlaManager#minRequiredInstances}.
+ */
+ @Test
+ public void testCheckCountSlaFailsDueToMinRequiredInstancesActs() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1));
+
+ // expect that the fetchTask in the work is called, after sla check passes
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ ISlaPolicy.build(
+ SlaPolicy.countSlaPolicy(
+ new CountSlaPolicy()
+ .setCount(0)
+ .setDurationSecs(0))),
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that SLA check is skipped and the supplied {@link Storage.MutateWork} gets
+ * executed for a job is not the correct tier.
+ */
+ @Test
+ public void testCheckCountSlaFailsDueToTierActs() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING, 1000, false);
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1));
+
+ // expect that the fetchTask in the work is called, after sla check passes
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ ISlaPolicy.build(
+ SlaPolicy.countSlaPolicy(
+ new CountSlaPolicy()
+ .setCount(0)
+ .setDurationSecs(0))),
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that SLA check fails and the supplied {@link Storage.MutateWork} does not get
+ * executed for a job that has {@link CountSlaPolicy#count} tasks that have been RUNNING for
+ * the required {@link CountSlaPolicy#durationSecs} and 1 task in RUNNING for less than
+ * {@link CountSlaPolicy#durationSecs} with {@link CountSlaPolicy#count} set to 2.
+ */
+ @Test
+ public void testCheckCountSlaFailsDueToRunningTimeDoesNotAct() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ IScheduledTask task2 = makeTask("taskB", 2, RUNNING);
+ IScheduledTask task3 = makeTask("taskC", 3, System.currentTimeMillis());
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // mock calls to fetch all RUNNING tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).byStatus(RUNNING)))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ COUNT_SLA_POLICY,
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that when {@code force} is {@code True} supplied {@link Storage.MutateWork} gets
+ * executed for a job that has 1 task in RUNNING for less than {@link CountSlaPolicy#durationSecs}
+ * with {@link CountSlaPolicy#count} set to 2, without checking SLA.
+ */
+ @Test
+ public void testCheckCountSlaCheckForceAct() {
+ IScheduledTask task1 = makeTask("taskA", 1, System.currentTimeMillis());
+
+ // expect that the fetchTask is the work is called after force
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ COUNT_SLA_POLICY,
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ true);
+ }
+
+ /**
+ * Verifies that SLA check passes and the supplied {@link Storage.MutateWork} gets executed
+ * for a job that has more {@link PercentageSlaPolicy#percentage} tasks that have been RUNNING for
+ * the required {@link PercentageSlaPolicy#durationSecs} with
+ * {@link PercentageSlaPolicy#percentage} set to 66.
+ */
+ @Test
+ public void testCheckPercentageSlaPassesThenActs() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ IScheduledTask task2 = makeTask("taskB", 2, RUNNING);
+ IScheduledTask task3 = makeTask("taskC", 3, RUNNING);
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // mock calls to fetch all RUNNING tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).byStatus(RUNNING)))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // expect that the fetchTask in the work is called, after sla check passes
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ PERCENTAGE_SLA_POLICY,
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that SLA check overridden and the supplied {@link Storage.MutateWork} gets executed
+ * for a job that requires aggressive {@link PercentageSlaPolicy#percentage} tasks to be RUNNING
+ * for the required {@link PercentageSlaPolicy#durationSecs} with
+ * {@link PercentageSlaPolicy#percentage} set to 67.
+ */
+ @Test
+ public void testCheckPercentageSlaWithAggressivePercentagePassesThenActs() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ IScheduledTask task2 = makeTask("taskB", 2, RUNNING);
+ IScheduledTask task3 = makeTask("taskC", 3, RUNNING);
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // mock calls to fetch all RUNNING tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).byStatus(RUNNING)))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // expect that the fetchTask in the work is called, after sla check passes
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ control.replay();
+
+ // We set the required running percentage to 67 which will allow 0 instances to be in
+ // maintenance is overridden to allow exactly 1 instance to be in maintenance.
+ slaManager.checkSlaThenAct(
+ task1,
+ ISlaPolicy.build(
+ SlaPolicy.percentageSlaPolicy(
+ new PercentageSlaPolicy()
+ .setPercentage(67)
+ .setDurationSecs(1800))),
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that SLA check is overridden and the supplied {@link Storage.MutateWork} gets
+ * executed for a job that requires aggressive {@link PercentageSlaPolicy#percentage} tasks to
+ * be RUNNING for the required {@link PercentageSlaPolicy#durationSecs} with
+ * {@link PercentageSlaPolicy#percentage} set to 67.
+ */
+ @Test
+ public void testCheckPercentageSlaWithAggressivePercentageFailsAct() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ IScheduledTask task2 = makeTask("taskB", 2, RUNNING);
+ IScheduledTask task3 = makeTask("taskC", 3, PENDING);
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // mock calls to fetch all RUNNING tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).byStatus(RUNNING)))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ control.replay();
+
+ // We set the required running percentage to 67 which will allow 0 instances to be in
+ // maintenance is not overridden to allow 1 instance to be in maintenance.
+ slaManager.checkSlaThenAct(
+ task1,
+ ISlaPolicy.build(
+ SlaPolicy.percentageSlaPolicy(
+ new PercentageSlaPolicy()
+ .setPercentage(67)
+ .setDurationSecs(1800))),
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that SLA check fails and the supplied {@link Storage.MutateWork} does not get
+ * executed for a job that has {@link PercentageSlaPolicy#percentage} tasks that have been
+ * RUNNING for the required {@link PercentageSlaPolicy#durationSecs} and 1 task in PENDING with
+ * {@link PercentageSlaPolicy#percentage} set to 66.
+ */
+ @Test
+ public void testCheckPercentageSlaFailsDoesNotAct() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ IScheduledTask task2 = makeTask("taskB", 2, RUNNING);
+ IScheduledTask task3 = makeTask("taskC", 3, PENDING);
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // mock calls to fetch all RUNNING tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).byStatus(RUNNING)))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ PERCENTAGE_SLA_POLICY,
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that SLA check is skipped and the supplied {@link Storage.MutateWork} gets
+ * executed for a job that does not meet {@link SlaManager#minRequiredInstances}.
+ */
+ @Test
+ public void testCheckPercentageSlaFailsDueToMinRequiredInstancesActs() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1));
+
+ // expect that the fetchTask in the work is called, after sla check passes
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ ISlaPolicy.build(
+ SlaPolicy.percentageSlaPolicy(
+ new PercentageSlaPolicy()
+ .setPercentage(0)
+ .setDurationSecs(0))),
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that SLA check is skipped and the supplied {@link Storage.MutateWork} gets
+ * executed for a job is not correct tier.
+ */
+ @Test
+ public void testCheckPercentageSlaFailsDueToTierActs() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING, 1000, false);
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1));
+
+ // expect that the fetchTask in the work is called, after sla check passes
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ ISlaPolicy.build(
+ SlaPolicy.percentageSlaPolicy(
+ new PercentageSlaPolicy()
+ .setPercentage(0)
+ .setDurationSecs(0))),
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that SLA check fails and the supplied {@link Storage.MutateWork} does not get
+ * executed for a job that has {@link PercentageSlaPolicy#percentage} tasks that have been
+ * RUNNING for the required {@link PercentageSlaPolicy#durationSecs} and 1 task in RUNNING for
+ * less than {@link PercentageSlaPolicy#durationSecs} with {@link PercentageSlaPolicy#percentage}
+ * set to 66.
+ */
+ @Test
+ public void testCheckPercentageSlaFailsDueToRunningTimeDoesNotAct() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ IScheduledTask task2 = makeTask("taskB", 2, RUNNING);
+ IScheduledTask task3 = makeTask("taskC", 3, System.currentTimeMillis());
+
+ // mock calls to fetch all active tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).active()))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ // mock calls to fetch all RUNNING tasks for the job for sla calculation
+ expect(storageUtil.taskStore.fetchTasks(Query.jobScoped(Tasks.getJob(task1)).byStatus(RUNNING)))
+ .andReturn(ImmutableSet.of(task1, task2, task3));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ PERCENTAGE_SLA_POLICY,
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ false);
+ }
+
+ /**
+ * Verifies that when {@code force} is {@code True} the supplied {@link Storage.MutateWork} gets
+ * executed for a job that has 1 task in RUNNING for less than
+ * {@link PercentageSlaPolicy#durationSecs}
+ * with {@link PercentageSlaPolicy#percentage} set to 66, without checking SLA.
+ */
+ @Test
+ public void testCheckPercentageSlaCheckForceAct() {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+
+ // expect that the fetchTask is the work is called after force
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ PERCENTAGE_SLA_POLICY,
+ storeProvider -> storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId()),
+ true);
+ }
+
+ /**
+ * Verifies that SLA check passes and the supplied {@link Storage.MutateWork} gets executed
+ * for a job when {@link CoordinatorSlaPolicy#coordinatorUrl} responds
+ * with {@code {"drain": true}}.
+ */
+ @Test
+ public void testCheckCoordinatorSlaPassesThenActs() throws Exception {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ CountDownLatch workCalled = new CountDownLatch(1);
+
+ jettyServer.setHandler(mockCoordinatorResponse(task1, "{\"drain\": true}"));
+ jettyServer.start();
+
+ // expect that the fetchTask in the work is called, after sla check passes
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ control.replay();
+
+ while (!coordinatorResponded.await(100, TimeUnit.MILLISECONDS)) {
+ slaManager.checkSlaThenAct(
+ task1,
+ createCoordinatorSlaPolicy(),
+ storeProvider -> {
+ // set the marker to indicate that we performed the work
+ workCalled.countDown();
+ storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId());
+ return null;
+ },
+ false);
+ }
+
+ // wait until we are sure that the server has responded
+ coordinatorResponded.await();
+ workCalled.await();
+
+ assertEquals(0, coordinatorResponded.getCount());
+ // check the work was called
+ assertEquals(0, workCalled.getCount());
+ }
+
+ /**
+ * Verifies that SLA check passes and the supplied {@link Storage.MutateWork} gets executed
+ * for a job when {@link CoordinatorSlaPolicy#coordinatorUrl} responds
+ * with {@code {"drain": true}}.
+ */
+ @Test
+ public void testCheckCoordinatorSlaPassesThenActsCustomStatusKey() throws Exception {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ CountDownLatch workCalled = new CountDownLatch(1);
+
+ jettyServer.setHandler(mockCoordinatorResponse(task1, "{\"custom-key\": true}"));
+ jettyServer.start();
+
+ // expect that the fetchTask in the work is called, after sla check passes
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ control.replay();
+
+ while (!coordinatorResponded.await(100, TimeUnit.MILLISECONDS)) {
+ slaManager.checkSlaThenAct(
+ task1,
+ createCoordinatorSlaPolicy("custom-key"),
+ storeProvider -> {
+ // set the marker to indicate that we performed the work
+ workCalled.countDown();
+ storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId());
+ return null;
+ },
+ false);
+ }
+
+ workCalled.await();
+
+ assertEquals(0, coordinatorResponded.getCount());
+ // check the work was called
+ assertEquals(0, workCalled.getCount());
+ }
+
+ /**
+ * Verifies that SLA check passes and the supplied {@link Storage.MutateWork} does not get
+ * executed for a job when {@link CoordinatorSlaPolicy#coordinatorUrl} responds
+ * with {@code {"drain": false}}.
+ */
+ @Test
+ public void testCheckCoordinatorSlaFailsDoesNotAct() throws Exception {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ CountDownLatch workCalled = new CountDownLatch(1);
+
+ jettyServer.setHandler(mockCoordinatorResponse(task1, "{\"drain\": false}"));
+ jettyServer.start();
+
+ control.replay();
+
+ while (!coordinatorResponded.await(100, TimeUnit.MILLISECONDS)) {
+ slaManager.checkSlaThenAct(
+ task1,
+ createCoordinatorSlaPolicy(),
+ storeProvider -> {
+ // set the marker to indicate that we performed the work
+ workCalled.countDown();
+ storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId());
+ return null;
+ },
+ false);
+ }
+
+ try {
+ workCalled.await(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ // expected
+ }
+
+ assertEquals(0, coordinatorResponded.getCount());
+ // check the work was not called
+ assertEquals(1, workCalled.getCount());
+ }
+
+ /**
+ * Verifies that when {@code force} is {@code True} the supplied {@link Storage.MutateWork} gets
+ * executed without checking the SLA.
+ */
+ @Test
+ public void testCheckCoordinatorSlaCheckForceAct() throws Exception {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ CountDownLatch workCalled = new CountDownLatch(1);
+
+ jettyServer.setHandler(mockCoordinatorResponse(task1, "{\"drain\": false}"));
+ jettyServer.start();
+
+ // expect that the fetchTask in the work is called after force
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ createCoordinatorSlaPolicy(),
+ storeProvider -> {
+ // set the marker to indicate that we performed the work
+ workCalled.countDown();
+ storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId());
+ return null;
+ },
+ true);
+
+ workCalled.await();
+
+ // coordinator is not contacted
+ assertEquals(1, coordinatorResponded.getCount());
+ // check the work was called
+ assertEquals(0, workCalled.getCount());
+ }
+
+ /**
+ * Verifies that SLA check fails and the supplied {@link Storage.MutateWork} does not get
+ * executed for a job when {@link CoordinatorSlaPolicy#coordinatorUrl} responds
+ * with error.
+ */
+ @Test
+ public void testCheckCoordinatorSlaErrorDoesNotAct() throws Exception {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ CountDownLatch workCalled = new CountDownLatch(1);
+
+ jettyServer.setHandler(mockCoordinatorError());
+ jettyServer.start();
+
+ control.replay();
+
+ while (!coordinatorResponded.await(100, TimeUnit.MILLISECONDS)) {
+ slaManager.checkSlaThenAct(
+ task1,
+ createCoordinatorSlaPolicy(),
+ storeProvider -> {
+ // set the marker to indicate that we performed the work
+ workCalled.countDown();
+ storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId());
+ return null;
+ },
+ false);
+ }
+
+ try {
+ workCalled.await(1, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ // expected
+ }
+
+ assertEquals(0, coordinatorResponded.getCount());
+ // check the work was not called
+ assertEquals(1, workCalled.getCount());
+ }
+
+ /**
+ * Verifies that SLA check fails and the supplied {@link Storage.MutateWork} does not get
+ * executed for a job when {@link CoordinatorSlaPolicy#coordinatorUrl} throws.
+ */
+ @Test
+ public void testCheckCoordinatorSlaThrowsDoesNotAct() throws Exception {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ CountDownLatch workCalled = new CountDownLatch(1);
+
+ jettyServer.setHandler(mockCoordinatorError());
+ jettyServer.start();
+
+ control.replay();
+
+ slaManager.checkSlaThenAct(
+ task1,
+ ISlaPolicy.build(SlaPolicy.coordinatorSlaPolicy(
+ new CoordinatorSlaPolicy()
+ .setCoordinatorUrl("http://does.not.exist.com"))),
+ storeProvider -> {
+ // set the marker to indicate that we performed the work
+ workCalled.countDown();
+ storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId());
+ return null;
+ },
+ false);
+
+ try {
+ workCalled.await(1, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ // expected
+ }
+
+ // check the work was not called
+ assertEquals(1, workCalled.getCount());
+ }
+
+ /**
+ * Start 2 actions simultaneously and make the first one to enter the coordinator lock to block
+ * on a semaphore. Only when the semaphore is released will the next action be able acquire the
+ * lock. At the end both actions should have completed.
+ */
+ @Test
+ public void testCheckCoordinatorSlaNoLockDoesNotAct() throws Exception {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ IScheduledTask task2 = makeTask("taskB", 2, RUNNING);
+
+ CountDownLatch blocked1 = new CountDownLatch(1);
+ CountDownLatch blocked2 = new CountDownLatch(1);
+ CountDownLatch action1 = new CountDownLatch(1);
+ CountDownLatch action2 = new CountDownLatch(1);
+ CountDownLatch finishAction1 = new CountDownLatch(1);
+ CountDownLatch finishAction2 = new CountDownLatch(1);
+
+ jettyServer.setHandler(mockCoordinatorResponses("{\"drain\": true}"));
+ jettyServer.start();
+
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ expect(storageUtil.taskStore.fetchTask(task2.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task2));
+
+ control.replay();
+
+ // kick off a new thread to attempt starting action1
+ new Thread(() -> {
+ try {
+ // wait until action1 enters the lock
+ while (!action1.await(100, TimeUnit.MILLISECONDS)) {
+ // start action1
+ slaManager.checkSlaThenAct(
+ task1,
+ createCoordinatorSlaPolicy(),
+ storeProvider -> {
+ LOG.info("Starting action1 for task:{}", slaManager.getTaskKey(task1));
+ // set the marker to indicate that we started the work
+ action1.countDown();
+ storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId());
+ // we will block here to make sure that the lock is held
+ blocked1.await();
+ finishAction1.countDown();
+ LOG.info("Finished action1 for task:{}", slaManager.getTaskKey(task1));
+ return null;
+ },
+ false);
+ }
+ } catch (InterruptedException e) {
+ fail();
+ }
+ }).start();
+
+ // wait for action to ge the lock
+ action1.await();
+
+ // kick off a new thread to attempt starting action2
+ new Thread(() -> {
+ try {
+ // wait until action2 enters the lock
+ while (!action2.await(100, TimeUnit.MILLISECONDS)) {
+ // start action2
+ slaManager.checkSlaThenAct(
+ task2,
+ createCoordinatorSlaPolicy(),
+ storeProvider -> {
+ LOG.info("Starting action2 for task:{}", slaManager.getTaskKey(task2));
+ // set the marker to indicate that we started the work
+ action2.countDown();
+ storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task2.getAssignedTask().getTaskId());
+ // we will block here to make sure that the lock is held
+ blocked2.await();
+ finishAction2.countDown();
+ LOG.info("Finished action2 for task:{}", slaManager.getTaskKey(task2));
+ return null;
+ },
+ false);
+ }
+ } catch (InterruptedException e) {
+ fail();
+ }
+ }).start();
+
+ // both actions should not have entered critical section
+ assertNotEquals(action1.getCount(), action2.getCount());
+ // action1 should have entered the lock
+ assertEquals(0, action1.getCount());
+ // action2 should not have entered the lock
+ assertEquals(1, action2.getCount());
+
+ // unblock blocked action1
+ blocked1.countDown();
+ // wait for both of the actions to finish
+ finishAction1.await();
+
+ // wait for the second action to enter the lock
+ action2.await();
+
+ // action2 should have entered the lock
+ assertEquals(0, action2.getCount());
+
+ // unblock blocked action2
+ blocked2.countDown();
+
+ finishAction2.await();
+ }
+
+ /**
+ * Start 2 actions simultaneously with different coordinator urls, so they can both enter the
+ * appropriate coordinator locks and try to acquire the semaphore. Both the actions must be able
+ * to enter the critical area and start the work simultaneously.
+ * At the end only one action would have fully completed the work, since the semaphore is not
+ * released.
+ */
+ @Test
+ public void testCheckCoordinatorSlaParallelAct() throws Exception {
+ IScheduledTask task1 = makeTask("taskA", 1, RUNNING);
+ IScheduledTask task2 = makeTask("taskB", 2, RUNNING);
+
+ Semaphore blocked = new Semaphore(1);
+ CountDownLatch action1 = new CountDownLatch(1);
+ CountDownLatch action2 = new CountDownLatch(1);
+ CountDownLatch finished = new CountDownLatch(2);
+
+ jettyServer.setHandler(mockCoordinatorResponses("{\"drain\": true}"));
+ jettyServer.start();
+
+ expect(storageUtil.taskStore.fetchTask(task1.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task1));
+
+ expect(storageUtil.taskStore.fetchTask(task2.getAssignedTask().getTaskId()))
+ .andReturn(Optional.of(task2));
+
+ control.replay();
+
+ // kick off a new thread to attempt starting action1
+ new Thread(() -> {
+ try {
+ // wait until action1 enters the lock
+ while (!action1.await(100, TimeUnit.MILLISECONDS)) {
+ // start action1
+ slaManager.checkSlaThenAct(
+ task1,
+ ISlaPolicy.build(SlaPolicy.coordinatorSlaPolicy(
+ new CoordinatorSlaPolicy()
+ .setCoordinatorUrl(
+ // Note that the url is different although referring to the same server
+ String.format("http://localhost:%d", jettyServer.getURI().getPort())))),
+ storeProvider -> {
+ LOG.info("Starting action for task:{}", slaManager.getTaskKey(task1));
+ // set the marker to indicate that we started the work
+ action1.countDown();
+ storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task1.getAssignedTask().getTaskId());
+ // we will block here to make sure that the lock is held
+ blocked.acquire();
+ finished.countDown();
+ LOG.info("Finished action for task:{}", slaManager.getTaskKey(task1));
+ return null;
+ },
+ false);
+ }
+ } catch (InterruptedException e) {
+ fail();
+ }
+ }).start();
+
+ // kick off a new thread to attempt starting action2
+ new Thread(() -> {
+ try {
+ // wait until action2 enters the lock
+ while (!action2.await(100, TimeUnit.MILLISECONDS)) {
+ // start action2
+ slaManager.checkSlaThenAct(
+ task2,
+ ISlaPolicy.build(SlaPolicy.coordinatorSlaPolicy(
+ new CoordinatorSlaPolicy()
+ .setCoordinatorUrl(
+ // Note that the url is different although referring to the same server
+ String.format("http://0.0.0.0:%d", jettyServer.getURI().getPort())))),
+ storeProvider -> {
+ LOG.info("Starting action for task:{}", slaManager.getTaskKey(task2));
+ // set the marker to indicate that we started the work
+ action2.countDown();
+ storeProvider
+ .getUnsafeTaskStore()
+ .fetchTask(task2.getAssignedTask().getTaskId());
+ // we will block here to make sure that the lock is held
+ blocked.acquire();
+ finished.countDown();
+ LOG.info("Finished action for task:{}", slaManager.getTaskKey(task2));
+ return null;
+ },
+ false);
+ }
+ } catch (InterruptedException e) {
+ fail();
+ }
+ }).start();
+
+ // wait for both of the actions to get the lock
+ action1.await();
+ action2.await();
+
+ assertEquals(0, action1.getCount());
+ assertEquals(0, action2.getCount());
+
+ // only 1 action has fully completed
+ assertEquals(1, finished.getCount());
+
+ // unblock the blocked action
+ blocked.release();
+
+ finished.await();
+
+ // both actions have fully completed
+ assertEquals(0, finished.getCount());
+ }
+
+ private ISlaPolicy createCoordinatorSlaPolicy() {
+ return createCoordinatorSlaPolicy("drain");
+ }
+
+ private ISlaPolicy createCoordinatorSlaPolicy(String statusKey) {
+ return ISlaPolicy.build(SlaPolicy.coordinatorSlaPolicy(
+ new CoordinatorSlaPolicy()
+ .setCoordinatorUrl(String.format("http://localhost:%d", jettyServer.getURI().getPort()))
+ .setStatusKey(statusKey)
+ ));
+ }
+
+ private AbstractHandler mockCoordinatorResponse(
+ IScheduledTask task,
+ String pollResponse) {
+
+ return new AbstractHandler() {
+ @Override
+ public void handle(
+ String target,
+ Request baseRequest,
+ HttpServletRequest request,
+ HttpServletResponse response) throws IOException {
+ try {
+ String taskKey = slaManager.getTaskKey(task);
+ String query = Joiner
+ .on("=")
+ .join(SlaManager.TASK_PARAM, URLEncoder.encode(taskKey, "UTF-8"));
+ String body = new TSerializer(new TSimpleJSONProtocol.Factory())
+ .toString(task.newBuilder());
+ if (request.getQueryString().equals(query)
+ && request.getReader().lines().collect(Collectors.joining()).equals(body)) {
+ createResponse(baseRequest, response, pollResponse);
+ }
+ coordinatorResponded.countDown();
+ } catch (TException e) {
+ fail();
+ }
+ }
+ };
+ }
+
+ private AbstractHandler mockCoordinatorResponses(String mockResponse) {
+ return new AbstractHandler() {
+ @Override
+ public void handle(String target, Request baseRequest, HttpServletRequest request,
+ HttpServletResponse response) throws IOException {
+ createResponse(baseRequest, response, mockResponse);
+ }
+ };
+ }
+
+ private void createResponse(
+ Request baseRequest,
+ HttpServletResponse response,
+ String mockResponse) throws IOException {
+ response.setStatus(HttpServletResponse.SC_OK);
+ response.getWriter().write(mockResponse);
+ response.getWriter().flush();
+ baseRequest.setHandled(true);
+ }
+
+ private AbstractHandler mockCoordinatorError() {
+ return new AbstractHandler() {
+ @Override
+ public void handle(String target, Request baseRequest, HttpServletRequest request,
+ HttpServletResponse response) {
+ response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
+ baseRequest.setHandled(true);
+ coordinatorResponded.countDown();
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/sla/SlaModuleTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/sla/SlaModuleTest.java b/src/test/java/org/apache/aurora/scheduler/sla/SlaModuleTest.java
index d37e7a0..b38a1e3 100644
--- a/src/test/java/org/apache/aurora/scheduler/sla/SlaModuleTest.java
+++ b/src/test/java/org/apache/aurora/scheduler/sla/SlaModuleTest.java
@@ -30,17 +30,21 @@ import org.apache.aurora.common.stats.StatsProvider;
import org.apache.aurora.common.testing.easymock.EasyMockTest;
import org.apache.aurora.common.util.Clock;
import org.apache.aurora.common.util.testing.FakeClock;
+import org.apache.aurora.gen.ServerInfo;
+import org.apache.aurora.scheduler.TierManager;
import org.apache.aurora.scheduler.app.LifecycleModule;
import org.apache.aurora.scheduler.base.Query;
import org.apache.aurora.scheduler.config.types.TimeAmount;
import org.apache.aurora.scheduler.sla.SlaModule.SlaUpdater;
import org.apache.aurora.scheduler.storage.Storage;
+import org.apache.aurora.scheduler.storage.entities.IServerInfo;
import org.apache.aurora.scheduler.storage.testing.StorageTestUtil;
import org.easymock.EasyMock;
import org.junit.Before;
import org.junit.Test;
import static org.apache.aurora.gen.ScheduleStatus.PENDING;
+import static org.apache.aurora.scheduler.base.TaskTestUtil.TIER_MANAGER;
import static org.apache.aurora.scheduler.sla.MetricCalculator.MetricCategory.JOB_UPTIMES;
import static org.apache.aurora.scheduler.sla.MetricCalculator.MetricCategory.MEDIANS;
import static org.apache.aurora.scheduler.sla.MetricCalculator.MetricCategory.PLATFORM_UPTIME;
@@ -50,6 +54,9 @@ import static org.junit.Assert.assertNotNull;
public class SlaModuleTest extends EasyMockTest {
+ private static final String CLUSTER_NAME = "test_cluster";
+ private static final String STATS_URL_PREFIX = "fake_url";
+
private Injector injector;
private FakeClock clock;
private StorageTestUtil storageUtil;
@@ -57,7 +64,7 @@ public class SlaModuleTest extends EasyMockTest {
private SlaModule module;
@Before
- public void setUp() throws Exception {
+ public void setUp() {
storageUtil = new StorageTestUtil(this);
clock = new FakeClock();
statsProvider = createMock(StatsProvider.class);
@@ -76,6 +83,12 @@ public class SlaModuleTest extends EasyMockTest {
bind(Clock.class).toInstance(clock);
bind(Storage.class).toInstance(storageUtil.storage);
bind(StatsProvider.class).toInstance(statsProvider);
+ bind(TierManager.class).toInstance(TIER_MANAGER);
+ bind(IServerInfo.class).toInstance(
+ IServerInfo.build(
+ new ServerInfo()
+ .setClusterName(CLUSTER_NAME)
+ .setStatsUrlPrefix(STATS_URL_PREFIX)));
}
}).build()
);
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/state/MaintenanceControllerImplTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/state/MaintenanceControllerImplTest.java b/src/test/java/org/apache/aurora/scheduler/state/MaintenanceControllerImplTest.java
deleted file mode 100644
index 770846e..0000000
--- a/src/test/java/org/apache/aurora/scheduler/state/MaintenanceControllerImplTest.java
+++ /dev/null
@@ -1,264 +0,0 @@
-/**
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.aurora.scheduler.state;
-
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.Executor;
-
-import com.google.common.collect.ImmutableSet;
-import com.google.common.util.concurrent.MoreExecutors;
-import com.google.inject.AbstractModule;
-import com.google.inject.Guice;
-import com.google.inject.Injector;
-
-import org.apache.aurora.common.quantity.Amount;
-import org.apache.aurora.common.quantity.Time;
-import org.apache.aurora.common.stats.StatsProvider;
-import org.apache.aurora.common.testing.easymock.EasyMockTest;
-import org.apache.aurora.gen.HostAttributes;
-import org.apache.aurora.gen.HostStatus;
-import org.apache.aurora.gen.MaintenanceMode;
-import org.apache.aurora.gen.ScheduleStatus;
-import org.apache.aurora.gen.ScheduledTask;
-import org.apache.aurora.scheduler.SchedulerModule.TaskEventBatchWorker;
-import org.apache.aurora.scheduler.async.AsyncModule.AsyncExecutor;
-import org.apache.aurora.scheduler.base.Query;
-import org.apache.aurora.scheduler.base.TaskTestUtil;
-import org.apache.aurora.scheduler.base.Tasks;
-import org.apache.aurora.scheduler.events.EventSink;
-import org.apache.aurora.scheduler.events.PubsubEvent.TaskStateChange;
-import org.apache.aurora.scheduler.events.PubsubEventModule;
-import org.apache.aurora.scheduler.storage.Storage;
-import org.apache.aurora.scheduler.storage.entities.IHostAttributes;
-import org.apache.aurora.scheduler.storage.entities.IHostStatus;
-import org.apache.aurora.scheduler.storage.entities.IScheduledTask;
-import org.apache.aurora.scheduler.storage.testing.StorageTestUtil;
-import org.apache.aurora.scheduler.testing.FakeStatsProvider;
-import org.apache.mesos.v1.Protos;
-import org.junit.Before;
-import org.junit.Test;
-
-import static org.apache.aurora.gen.MaintenanceMode.DRAINED;
-import static org.apache.aurora.gen.MaintenanceMode.DRAINING;
-import static org.apache.aurora.gen.MaintenanceMode.NONE;
-import static org.apache.aurora.gen.MaintenanceMode.SCHEDULED;
-import static org.apache.aurora.gen.ScheduleStatus.KILLED;
-import static org.apache.aurora.gen.ScheduleStatus.RUNNING;
-import static org.apache.aurora.scheduler.state.MaintenanceController.MaintenanceControllerImpl;
-import static org.apache.aurora.scheduler.testing.BatchWorkerUtil.expectBatchExecute;
-import static org.easymock.EasyMock.expect;
-import static org.junit.Assert.assertEquals;
-
-public class MaintenanceControllerImplTest extends EasyMockTest {
-
- private static final String HOST_A = "a";
- private static final Set<String> A = ImmutableSet.of(HOST_A);
- private static final Protos.OfferID OFFER_ID = Protos.OfferID.newBuilder()
- .setValue("offer-id")
- .build();
- private static final Protos.AgentID AGENT_ID = Protos.AgentID.newBuilder()
- .setValue("agent-id")
- .build();
- private static final Protos.FrameworkID FRAMEWORK_ID = Protos.FrameworkID.newBuilder()
- .setValue("framework-id")
- .build();
- private static final Protos.URL AGENT_URL = Protos.URL.newBuilder()
- .setAddress(Protos.Address.newBuilder()
- .setHostname(HOST_A)
- .setPort(5051))
- .setScheme("http")
- .build();
- private static final Protos.Unavailability UNAVAILABILITY = Protos.Unavailability.newBuilder()
- .setStart(Protos.TimeInfo.newBuilder()
- .setNanoseconds(Amount.of(1L, Time.MINUTES).as(Time.NANOSECONDS)))
- .build();
-
- private static final Protos.InverseOffer INVERSE_OFFER = Protos.InverseOffer.newBuilder()
- .setId(OFFER_ID)
- .setAgentId(AGENT_ID)
- .setUrl(AGENT_URL)
- .setFrameworkId(FRAMEWORK_ID)
- .setUnavailability(UNAVAILABILITY)
- .build();
-
- private StorageTestUtil storageUtil;
- private StateManager stateManager;
- private MaintenanceController maintenance;
- private EventSink eventSink;
-
- @Before
- public void setUp() throws Exception {
- storageUtil = new StorageTestUtil(this);
- storageUtil.expectOperations();
- stateManager = createMock(StateManager.class);
- TaskEventBatchWorker batchWorker = createMock(TaskEventBatchWorker.class);
- expectBatchExecute(batchWorker, storageUtil.storage, control).anyTimes();
-
- Injector injector = Guice.createInjector(
- new PubsubEventModule(),
- new AbstractModule() {
- @Override
- protected void configure() {
- StateModule.bindMaintenanceController(binder());
- bind(Storage.class).toInstance(storageUtil.storage);
- bind(StateManager.class).toInstance(stateManager);
- bind(StatsProvider.class).toInstance(new FakeStatsProvider());
- bind(Executor.class).annotatedWith(AsyncExecutor.class)
- .toInstance(MoreExecutors.directExecutor());
- bind(TaskEventBatchWorker.class).toInstance(batchWorker);
- }
- });
- maintenance = injector.getInstance(MaintenanceController.class);
- eventSink = PubsubTestUtil.startPubsub(injector);
- }
-
- private static IScheduledTask makeTask(String host, String taskId) {
- ScheduledTask builder = TaskTestUtil.addStateTransition(
- TaskTestUtil.makeTask(taskId, TaskTestUtil.JOB),
- RUNNING,
- 1000).newBuilder();
- builder.getAssignedTask().setSlaveHost(host);
- return IScheduledTask.build(builder);
- }
-
- @Test
- public void testMaintenanceCycle() {
- IScheduledTask task1 = makeTask(HOST_A, "taskA");
- IScheduledTask task2 = makeTask(HOST_A, "taskB");
-
- expectMaintenanceModeChange(HOST_A, SCHEDULED);
- expectFetchTasksByHost(HOST_A, ImmutableSet.of(task1, task2));
- expectTaskDraining(task1);
- expectTaskDraining(task2);
- expectMaintenanceModeChange(HOST_A, DRAINING);
- IHostAttributes attributes =
- IHostAttributes.build(new HostAttributes().setHost(HOST_A).setMode(DRAINING));
-
- expect(storageUtil.attributeStore.getHostAttributes(HOST_A))
- .andReturn(Optional.of(attributes)).times(2);
-
- expect(storageUtil.attributeStore.getHostAttributes()).andReturn(ImmutableSet.of(attributes));
- expectFetchTasksByHost(HOST_A, ImmutableSet.of(task2));
- // TaskA is KILLED and therefore no longer active
- expectFetchTasksByHost(HOST_A, ImmutableSet.of());
- expectMaintenanceModeChange(HOST_A, DRAINED);
- expectMaintenanceModeChange(HOST_A, NONE);
-
- control.replay();
-
- assertStatus(HOST_A, SCHEDULED, maintenance.startMaintenance(A));
- assertStatus(HOST_A, DRAINING, maintenance.drain(A));
- assertStatus(HOST_A, DRAINING, maintenance.getStatus(A));
- eventSink.post(
- TaskStateChange.transition(
- IScheduledTask.build(task1.newBuilder().setStatus(KILLED)), RUNNING));
- eventSink.post(
- TaskStateChange.transition(
- IScheduledTask.build(task2.newBuilder().setStatus(KILLED)), RUNNING));
- assertStatus(HOST_A, NONE, maintenance.endMaintenance(A));
- }
-
- @Test
- public void testUnknownHost() {
- expect(storageUtil.attributeStore.getHostAttributes("b"))
- .andReturn(Optional.empty());
-
- control.replay();
-
- assertEquals(ImmutableSet.of(),
- maintenance.startMaintenance(ImmutableSet.of("b")));
- }
-
- @Test
- public void testDrainEmptyHost() {
- expectMaintenanceModeChange(HOST_A, SCHEDULED);
- expectFetchTasksByHost(HOST_A, ImmutableSet.of());
- expectMaintenanceModeChange(HOST_A, DRAINED);
-
- control.replay();
-
- assertStatus(HOST_A, SCHEDULED, maintenance.startMaintenance(A));
- assertStatus(HOST_A, DRAINED, maintenance.drain(A));
- }
-
- @Test
- public void testEndEarly() {
- expectMaintenanceModeChange(HOST_A, SCHEDULED);
- expectMaintenanceModeChange(HOST_A, NONE);
- expect(storageUtil.attributeStore.getHostAttributes(HOST_A)).andReturn(Optional.of(
- IHostAttributes.build(new HostAttributes().setHost(HOST_A).setMode(NONE))));
-
- control.replay();
-
- assertStatus(HOST_A, SCHEDULED, maintenance.startMaintenance(A));
-
- // End maintenance without DRAINING.
- assertStatus(HOST_A, NONE, maintenance.endMaintenance(A));
-
- // Make sure a later transition on the host does not cause any ill effects that could surface
- // from stale internal state.
- eventSink.post(TaskStateChange.transition(
- IScheduledTask.build(makeTask(HOST_A, "taskA").newBuilder().setStatus(KILLED)), RUNNING));
- }
-
- @Test
- public void testGetMode() {
- expect(storageUtil.attributeStore.getHostAttributes(HOST_A)).andReturn(Optional.of(
- IHostAttributes.build(new HostAttributes().setHost(HOST_A).setMode(DRAINING))));
- expect(storageUtil.attributeStore.getHostAttributes("unknown")).andReturn(Optional.empty());
-
- control.replay();
-
- assertEquals(DRAINING, maintenance.getMode(HOST_A));
- assertEquals(NONE, maintenance.getMode("unknown"));
- }
-
- @Test
- public void testInverseOfferDrain() {
- IScheduledTask task1 = makeTask(HOST_A, "taskA");
- expectFetchTasksByHost(HOST_A, ImmutableSet.of(task1));
- expectTaskDraining(task1);
-
- control.replay();
- maintenance.drainForInverseOffer(INVERSE_OFFER);
- }
-
- private void expectTaskDraining(IScheduledTask task) {
- expect(stateManager.changeState(
- storageUtil.mutableStoreProvider,
- Tasks.id(task),
- Optional.empty(),
- ScheduleStatus.DRAINING,
- MaintenanceControllerImpl.DRAINING_MESSAGE))
- .andReturn(StateChangeResult.SUCCESS);
- }
-
- private void expectFetchTasksByHost(String hostName, Set<IScheduledTask> tasks) {
- expect(storageUtil.taskStore.fetchTasks(Query.slaveScoped(hostName).active())).andReturn(tasks);
- }
-
- private void expectMaintenanceModeChange(String hostName, MaintenanceMode mode) {
- IHostAttributes attributes = IHostAttributes.build(new HostAttributes().setHost(hostName));
-
- expect(storageUtil.attributeStore.getHostAttributes(hostName))
- .andReturn(Optional.of(attributes));
- IHostAttributes updated = IHostAttributes.build(attributes.newBuilder().setMode(mode));
- expect(storageUtil.attributeStore.saveHostAttributes(updated)).andReturn(true);
- }
-
- private void assertStatus(String host, MaintenanceMode mode, Set<IHostStatus> statuses) {
- assertEquals(ImmutableSet.of(IHostStatus.build(new HostStatus(host, mode))), statuses);
- }
-}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterfaceTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterfaceTest.java b/src/test/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterfaceTest.java
index 2cf66d8..0fc3673 100644
--- a/src/test/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterfaceTest.java
+++ b/src/test/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterfaceTest.java
@@ -50,6 +50,7 @@ import org.apache.aurora.gen.ListBackupsResult;
import org.apache.aurora.gen.MaintenanceMode;
import org.apache.aurora.gen.MesosContainer;
import org.apache.aurora.gen.Metadata;
+import org.apache.aurora.gen.PercentageSlaPolicy;
import org.apache.aurora.gen.PulseJobUpdateResult;
import org.apache.aurora.gen.QueryRecoveryResult;
import org.apache.aurora.gen.Range;
@@ -62,6 +63,7 @@ import org.apache.aurora.gen.ResponseDetail;
import org.apache.aurora.gen.Result;
import org.apache.aurora.gen.ScheduleStatus;
import org.apache.aurora.gen.ScheduledTask;
+import org.apache.aurora.gen.SlaPolicy;
import org.apache.aurora.gen.StartJobUpdateResult;
import org.apache.aurora.gen.TaskConfig;
import org.apache.aurora.gen.TaskConstraint;
@@ -77,10 +79,10 @@ import org.apache.aurora.scheduler.configuration.SanitizedConfiguration;
import org.apache.aurora.scheduler.cron.CronException;
import org.apache.aurora.scheduler.cron.CronJobManager;
import org.apache.aurora.scheduler.cron.SanitizedCronJob;
+import org.apache.aurora.scheduler.maintenance.MaintenanceController;
import org.apache.aurora.scheduler.quota.QuotaCheckResult;
import org.apache.aurora.scheduler.quota.QuotaManager;
import org.apache.aurora.scheduler.reconciliation.TaskReconciler;
-import org.apache.aurora.scheduler.state.MaintenanceController;
import org.apache.aurora.scheduler.state.StateChangeResult;
import org.apache.aurora.scheduler.state.StateManager;
import org.apache.aurora.scheduler.state.UUIDGenerator;
@@ -153,6 +155,7 @@ import static org.apache.aurora.scheduler.thrift.SchedulerThriftInterface.NOOP_J
import static org.apache.aurora.scheduler.thrift.SchedulerThriftInterface.NO_CRON;
import static org.apache.aurora.scheduler.thrift.SchedulerThriftInterface.PRUNE_TASKS;
import static org.apache.aurora.scheduler.thrift.SchedulerThriftInterface.RESTART_SHARDS;
+import static org.apache.aurora.scheduler.thrift.SchedulerThriftInterface.SLA_DRAIN_HOSTS;
import static org.apache.aurora.scheduler.thrift.SchedulerThriftInterface.START_JOB_UPDATE;
import static org.apache.aurora.scheduler.thrift.SchedulerThriftInterface.START_MAINTENANCE;
import static org.apache.aurora.scheduler.thrift.SchedulerThriftInterface.jobAlreadyExistsMessage;
@@ -1097,6 +1100,59 @@ public class SchedulerThriftInterfaceTest extends EasyMockTest {
assertEquals(1L, statsProvider.getLongValue(END_MAINTENANCE));
}
+ @Test
+ public void testSLAHostMaintenance() throws Exception {
+ Set<String> hostnames = ImmutableSet.of("a");
+ SlaPolicy defaultSlaPolicy = SlaPolicy.percentageSlaPolicy(
+ new PercentageSlaPolicy().setPercentage(95));
+ Set<IHostStatus> none = status("a", NONE);
+ Set<IHostStatus> scheduled = status("a", SCHEDULED);
+ Set<IHostStatus> draining = status("a", DRAINING);
+ Set<IHostStatus> drained = status("a", DRAINING);
+ expect(maintenance.getStatus(hostnames)).andReturn(none);
+ expect(maintenance.startMaintenance(hostnames)).andReturn(scheduled);
+ expect(maintenance.slaDrain(hostnames, defaultSlaPolicy, 10)).andReturn(draining);
+ expect(maintenance.getStatus(hostnames)).andReturn(draining);
+ expect(maintenance.getStatus(hostnames)).andReturn(drained);
+ expect(maintenance.endMaintenance(hostnames)).andReturn(none);
+
+ control.replay();
+
+ Hosts hosts = new Hosts(hostnames);
+
+ assertEquals(
+ IHostStatus.toBuildersSet(none),
+ thrift.maintenanceStatus(hosts).getResult().getMaintenanceStatusResult()
+ .getStatuses());
+ assertEquals(1L, statsProvider.getLongValue(MAINTENANCE_STATUS));
+ assertEquals(
+ IHostStatus.toBuildersSet(scheduled),
+ thrift.startMaintenance(hosts).getResult().getStartMaintenanceResult()
+ .getStatuses());
+ assertEquals(1L, statsProvider.getLongValue(START_MAINTENANCE));
+ assertEquals(
+ IHostStatus.toBuildersSet(draining),
+ thrift.slaDrainHosts(hosts, defaultSlaPolicy, 10)
+ .getResult()
+ .getDrainHostsResult()
+ .getStatuses());
+ assertEquals(1L, statsProvider.getLongValue(SLA_DRAIN_HOSTS));
+ assertEquals(
+ IHostStatus.toBuildersSet(draining),
+ thrift.maintenanceStatus(hosts).getResult().getMaintenanceStatusResult()
+ .getStatuses());
+ assertEquals(2L, statsProvider.getLongValue(MAINTENANCE_STATUS));
+ assertEquals(
+ IHostStatus.toBuildersSet(drained),
+ thrift.maintenanceStatus(hosts).getResult().getMaintenanceStatusResult()
+ .getStatuses());
+ assertEquals(3L, statsProvider.getLongValue(MAINTENANCE_STATUS));
+ assertEquals(
+ IHostStatus.toBuildersSet(none),
+ thrift.endMaintenance(hosts).getResult().getEndMaintenanceResult().getStatuses());
+ assertEquals(1L, statsProvider.getLongValue(END_MAINTENANCE));
+ }
+
private static Response okEmptyResponse() {
return response(OK, Optional.empty());
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/thrift/ThriftIT.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/thrift/ThriftIT.java b/src/test/java/org/apache/aurora/scheduler/thrift/ThriftIT.java
index 40851c4..f0199a4 100644
--- a/src/test/java/org/apache/aurora/scheduler/thrift/ThriftIT.java
+++ b/src/test/java/org/apache/aurora/scheduler/thrift/ThriftIT.java
@@ -25,9 +25,12 @@ import com.google.inject.Guice;
import com.google.inject.Injector;
import com.google.inject.Key;
import com.google.inject.Provides;
+import com.google.inject.TypeLiteral;
import org.apache.aurora.common.application.ShutdownStage;
import org.apache.aurora.common.base.Command;
+import org.apache.aurora.common.quantity.Amount;
+import org.apache.aurora.common.quantity.Time;
import org.apache.aurora.common.testing.easymock.EasyMockTest;
import org.apache.aurora.gen.AuroraAdmin;
import org.apache.aurora.gen.Container;
@@ -48,10 +51,12 @@ import org.apache.aurora.scheduler.app.ServiceGroupMonitor;
import org.apache.aurora.scheduler.app.local.FakeNonVolatileStorage;
import org.apache.aurora.scheduler.base.TaskTestUtil;
import org.apache.aurora.scheduler.config.CliOptions;
+import org.apache.aurora.scheduler.config.types.TimeAmount;
import org.apache.aurora.scheduler.configuration.ConfigurationManager;
import org.apache.aurora.scheduler.configuration.ConfigurationManager.ConfigurationManagerSettings;
import org.apache.aurora.scheduler.configuration.executor.ExecutorSettings;
import org.apache.aurora.scheduler.cron.quartz.CronModule;
+import org.apache.aurora.scheduler.maintenance.MaintenanceController;
import org.apache.aurora.scheduler.mesos.DriverFactory;
import org.apache.aurora.scheduler.mesos.DriverSettings;
import org.apache.aurora.scheduler.mesos.FrameworkInfoFactory;
@@ -138,6 +143,10 @@ public class ThriftIT extends EasyMockTest {
bindMock(StorageBackup.class);
bindMock(SnapshotStore.class);
bind(IServerInfo.class).toInstance(SERVER_INFO);
+ bind(new TypeLiteral<Amount<Long, Time>>() { })
+ .annotatedWith(
+ MaintenanceController.MaintenanceControllerImpl.PollingInterval.class)
+ .toInstance(new TimeAmount(1, Time.MINUTES));
}
@Provides
@@ -181,6 +190,8 @@ public class ThriftIT extends EasyMockTest {
true,
true,
false,
+ 20,
+ 1800,
ConfigurationManager.DEFAULT_ALLOWED_JOB_ENVIRONMENTS);
createThrift(configurationManagerSettings);
@@ -189,6 +200,7 @@ public class ThriftIT extends EasyMockTest {
TaskConfig task = TaskTestUtil.makeConfig(TaskTestUtil.JOB).newBuilder();
task.unsetExecutorConfig();
+ task.unsetSlaPolicy();
task.setProduction(false)
.setTier(TaskTestUtil.DEV_TIER_NAME)
.setContainer(Container.docker(new DockerContainer()
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/thrift/aop/MockDecoratedThrift.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/thrift/aop/MockDecoratedThrift.java b/src/test/java/org/apache/aurora/scheduler/thrift/aop/MockDecoratedThrift.java
index d412090..9903811 100644
--- a/src/test/java/org/apache/aurora/scheduler/thrift/aop/MockDecoratedThrift.java
+++ b/src/test/java/org/apache/aurora/scheduler/thrift/aop/MockDecoratedThrift.java
@@ -36,6 +36,7 @@ import org.apache.aurora.gen.JobUpdateRequest;
import org.apache.aurora.gen.ResourceAggregate;
import org.apache.aurora.gen.Response;
import org.apache.aurora.gen.ScheduleStatus;
+import org.apache.aurora.gen.SlaPolicy;
import org.apache.aurora.gen.TaskQuery;
import org.apache.aurora.scheduler.http.api.security.AuthorizingParam;
import org.apache.aurora.scheduler.thrift.auth.DecoratedThrift;
@@ -200,6 +201,12 @@ public class MockDecoratedThrift implements AnnotatedAuroraAdmin {
}
@Override
+ public Response slaDrainHosts(Hosts hosts, SlaPolicy defaultSlaPolicy, long timeoutSecs)
+ throws TException {
+ return this.annotatedAuroraAdmin.slaDrainHosts(hosts, defaultSlaPolicy, timeoutSecs);
+ }
+
+ @Override
public Response snapshot() throws TException {
return this.annotatedAuroraAdmin.snapshot();
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/python/apache/aurora/admin/test_maintenance.py
----------------------------------------------------------------------
diff --git a/src/test/python/apache/aurora/admin/test_maintenance.py b/src/test/python/apache/aurora/admin/test_maintenance.py
index ca0239b..d0f7956 100644
--- a/src/test/python/apache/aurora/admin/test_maintenance.py
+++ b/src/test/python/apache/aurora/admin/test_maintenance.py
@@ -18,7 +18,13 @@ from mock import Mock, patch
from twitter.common.contextutil import temporary_file
from apache.aurora.admin.host_maintenance import HostMaintenance
-from apache.aurora.admin.maintenance import host_activate, host_deactivate, host_drain, host_status
+from apache.aurora.admin.maintenance import (
+ host_activate,
+ host_deactivate,
+ host_drain,
+ host_status,
+ sla_host_drain
+)
from .util import AuroraClientCommandTest
@@ -29,12 +35,15 @@ from gen.apache.aurora.api.ttypes import (
HostStatus,
MaintenanceMode,
MaintenanceStatusResult,
+ PercentageSlaPolicy,
+ SlaPolicy,
StartMaintenanceResult
)
class TestMaintenanceCommands(AuroraClientCommandTest):
HOSTNAMES = ['us-grf-20', 'us-jim-47', 'us-suz-01']
+ DEFAULT_SLA_PERCENTAGE = 95
def make_mock_options(self):
mock_options = Mock()
@@ -46,6 +55,9 @@ class TestMaintenanceCommands(AuroraClientCommandTest):
mock_options.percentage = None
mock_options.duration = None
mock_options.reason = None
+ mock_options.default_duration = '30m'
+ mock_options.default_percentage = self.DEFAULT_SLA_PERCENTAGE
+ mock_options.timeout = '2h'
return mock_options
def create_host_statuses(self, maintenance_mode, skip_hosts=None):
@@ -127,9 +139,9 @@ class TestMaintenanceCommands(AuroraClientCommandTest):
mock_scheduler_proxy.startMaintenance.return_value = self.create_start_maintenance_result()
mock_scheduler_proxy.drainHosts.return_value = self.create_start_maintenance_result()
mock_vector = self.create_mock_probe_hosts_vector([
- self.create_probe_hosts(self.HOSTNAMES[0], 95, True, None),
- self.create_probe_hosts(self.HOSTNAMES[1], 95, True, None),
- self.create_probe_hosts(self.HOSTNAMES[2], 95, True, None)
+ self.create_probe_hosts(self.HOSTNAMES[0], self.DEFAULT_SLA_PERCENTAGE, True, None),
+ self.create_probe_hosts(self.HOSTNAMES[1], self.DEFAULT_SLA_PERCENTAGE, True, None),
+ self.create_probe_hosts(self.HOSTNAMES[2], self.DEFAULT_SLA_PERCENTAGE, True, None)
])
with contextlib.nested(
@@ -163,9 +175,9 @@ class TestMaintenanceCommands(AuroraClientCommandTest):
skip_hosts=['us-grf-20'])
mock_scheduler_proxy.drainHosts.return_value = self.create_start_maintenance_result()
mock_vector = self.create_mock_probe_hosts_vector([
- self.create_probe_hosts(self.HOSTNAMES[0], 95, True, None),
- self.create_probe_hosts(self.HOSTNAMES[1], 95, True, None),
- self.create_probe_hosts(self.HOSTNAMES[2], 95, True, None)
+ self.create_probe_hosts(self.HOSTNAMES[0], self.DEFAULT_SLA_PERCENTAGE, True, None),
+ self.create_probe_hosts(self.HOSTNAMES[1], self.DEFAULT_SLA_PERCENTAGE, True, None),
+ self.create_probe_hosts(self.HOSTNAMES[2], self.DEFAULT_SLA_PERCENTAGE, True, None)
])
with contextlib.nested(
@@ -192,9 +204,9 @@ class TestMaintenanceCommands(AuroraClientCommandTest):
mock_scheduler_proxy.startMaintenance.return_value = self.create_start_maintenance_result()
mock_scheduler_proxy.drainHosts.return_value = self.create_start_maintenance_result()
mock_vector = self.create_mock_probe_hosts_vector([
- self.create_probe_hosts(self.HOSTNAMES[0], 95, False, None),
- self.create_probe_hosts(self.HOSTNAMES[1], 95, False, None),
- self.create_probe_hosts(self.HOSTNAMES[2], 95, False, None)
+ self.create_probe_hosts(self.HOSTNAMES[0], self.DEFAULT_SLA_PERCENTAGE, False, None),
+ self.create_probe_hosts(self.HOSTNAMES[1], self.DEFAULT_SLA_PERCENTAGE, False, None),
+ self.create_probe_hosts(self.HOSTNAMES[2], self.DEFAULT_SLA_PERCENTAGE, False, None)
])
with contextlib.nested(
@@ -222,9 +234,9 @@ class TestMaintenanceCommands(AuroraClientCommandTest):
mock_scheduler_proxy.startMaintenance.return_value = self.create_start_maintenance_result()
mock_scheduler_proxy.drainHosts.return_value = self.create_start_maintenance_result()
mock_vector = self.create_mock_probe_hosts_vector([
- self.create_probe_hosts(self.HOSTNAMES[0], 95, False, None),
- self.create_probe_hosts(self.HOSTNAMES[1], 95, False, None),
- self.create_probe_hosts(self.HOSTNAMES[2], 95, False, None)
+ self.create_probe_hosts(self.HOSTNAMES[0], self.DEFAULT_SLA_PERCENTAGE, False, None),
+ self.create_probe_hosts(self.HOSTNAMES[1], self.DEFAULT_SLA_PERCENTAGE, False, None),
+ self.create_probe_hosts(self.HOSTNAMES[2], self.DEFAULT_SLA_PERCENTAGE, False, None)
])
mock_wait = Mock()
@@ -326,3 +338,46 @@ class TestMaintenanceCommands(AuroraClientCommandTest):
mock_scheduler_proxy.maintenanceStatus.assert_called_with(
Hosts(set(self.HOSTNAMES)),
retry=True)
+
+ def test_perform_sla_maintenance_hosts(self):
+ mock_options = self.make_mock_options()
+
+ mock_api, mock_scheduler_proxy = self.create_mock_api()
+ mock_scheduler_proxy.slaDrainHosts.return_value = self.create_start_maintenance_result()
+
+ with contextlib.nested(
+ patch('apache.aurora.client.api.SchedulerProxy', return_value=mock_scheduler_proxy),
+ patch('apache.aurora.admin.maintenance.CLUSTERS', new=self.TEST_CLUSTERS),
+ patch('twitter.common.app.get_options', return_value=mock_options)):
+ sla_host_drain([self.TEST_CLUSTER])
+
+ mock_scheduler_proxy.slaDrainHosts.assert_called_with(
+ Hosts(set(self.HOSTNAMES)),
+ SlaPolicy(percentageSlaPolicy=PercentageSlaPolicy(
+ percentage=self.DEFAULT_SLA_PERCENTAGE,
+ durationSecs=1800)),
+ 7200)
+ assert mock_scheduler_proxy.slaDrainHosts.call_count == 1
+
+ def test_perform_sla_maintenance_hosts_defaults(self):
+ mock_options = self.make_mock_options()
+ mock_options.default_percentage = self.DEFAULT_SLA_PERCENTAGE
+ mock_options.default_duration = '2m'
+ mock_options.timeout = '10m'
+
+ mock_api, mock_scheduler_proxy = self.create_mock_api()
+ mock_scheduler_proxy.slaDrainHosts.return_value = self.create_start_maintenance_result()
+
+ with contextlib.nested(
+ patch('apache.aurora.client.api.SchedulerProxy', return_value=mock_scheduler_proxy),
+ patch('apache.aurora.admin.maintenance.CLUSTERS', new=self.TEST_CLUSTERS),
+ patch('twitter.common.app.get_options', return_value=mock_options)):
+ sla_host_drain([self.TEST_CLUSTER])
+
+ mock_scheduler_proxy.slaDrainHosts.assert_called_with(
+ Hosts(set(self.HOSTNAMES)),
+ SlaPolicy(percentageSlaPolicy=PercentageSlaPolicy(
+ percentage=self.DEFAULT_SLA_PERCENTAGE,
+ durationSecs=120)),
+ 600)
+ assert mock_scheduler_proxy.slaDrainHosts.call_count == 1
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/python/apache/aurora/api_util.py
----------------------------------------------------------------------
diff --git a/src/test/python/apache/aurora/api_util.py b/src/test/python/apache/aurora/api_util.py
index 3fc9b47..2022db5 100644
--- a/src/test/python/apache/aurora/api_util.py
+++ b/src/test/python/apache/aurora/api_util.py
@@ -58,6 +58,9 @@ class SchedulerThriftApiSpec(ReadOnlyScheduler.Iface):
def drainHosts(self, hosts):
pass
+ def slaDrainHosts(self, hosts, defaultSlaPolicy, timeoutSecs):
+ pass
+
def maintenanceStatus(self, hosts):
pass
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/python/apache/aurora/client/api/test_scheduler_client.py
----------------------------------------------------------------------
diff --git a/src/test/python/apache/aurora/client/api/test_scheduler_client.py b/src/test/python/apache/aurora/client/api/test_scheduler_client.py
index f2a2eae..d1e1e43 100644
--- a/src/test/python/apache/aurora/client/api/test_scheduler_client.py
+++ b/src/test/python/apache/aurora/client/api/test_scheduler_client.py
@@ -45,6 +45,7 @@ from gen.apache.aurora.api.ttypes import (
ResponseCode,
ResponseDetail,
ScheduleStatus,
+ SlaPolicy,
TaskQuery
)
@@ -211,6 +212,14 @@ class TestSchedulerProxyAdminInjection(TestSchedulerProxyInjection):
self.mox.ReplayAll()
self.make_scheduler_proxy().drainHosts(Hosts())
+ def test_slaDrainHosts(self):
+ self.mock_thrift_client.slaDrainHosts(
+ IsA(Hosts),
+ IsA(SlaPolicy),
+ 10).AndReturn(DEFAULT_RESPONSE)
+ self.mox.ReplayAll()
+ self.make_scheduler_proxy().slaDrainHosts(Hosts(), SlaPolicy(), 10)
+
def test_maintenanceStatus(self):
self.mock_thrift_client.maintenanceStatus(
IsA(Hosts)).AndReturn(DEFAULT_RESPONSE)
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/python/apache/aurora/client/cli/test_add.py
----------------------------------------------------------------------
diff --git a/src/test/python/apache/aurora/client/cli/test_add.py b/src/test/python/apache/aurora/client/cli/test_add.py
index b22b9f7..ff4ccf6 100644
--- a/src/test/python/apache/aurora/client/cli/test_add.py
+++ b/src/test/python/apache/aurora/client/cli/test_add.py
@@ -20,7 +20,8 @@ from apache.aurora.client.cli.options import TaskInstanceKey
from .util import AuroraClientCommandTest, FakeAuroraCommandContext, mock_verb_options
-from gen.apache.aurora.api.ttypes import ScheduleStatus
+from gen.apache.aurora.api.constants import ACTIVE_STATES
+from gen.apache.aurora.api.ttypes import ScheduleStatus, TaskQuery
class TestAddCommand(AuroraClientCommandTest):
@@ -37,6 +38,8 @@ class TestAddCommand(AuroraClientCommandTest):
self._mock_options.open_browser = True
self._fake_context.add_expected_query_result(self.create_query_call_result(
self.create_scheduled_task(1, ScheduleStatus.RUNNING)))
+ self._fake_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
self._mock_api.add_instances.return_value = self.create_simple_success_response()
@@ -50,11 +53,17 @@ class TestAddCommand(AuroraClientCommandTest):
assert mock_webbrowser.mock_calls == [
call("http://something_or_other/scheduler/bozo/test/hello")
]
+ assert self._mock_api.query_no_configs.mock_calls == [
+ call(TaskQuery(jobKeys=[self.TEST_JOBKEY.to_thrift()], statuses=ACTIVE_STATES)),
+ call(TaskQuery(jobKeys=[self.TEST_JOBKEY.to_thrift()], statuses=ACTIVE_STATES))
+ ]
def test_wait_added_instances(self):
self._mock_options.wait_until = 'RUNNING'
self._fake_context.add_expected_query_result(self.create_query_call_result(
self.create_scheduled_task(1, ScheduleStatus.PENDING)))
+ self._fake_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
self._mock_api.add_instances.return_value = self.create_simple_success_response()
@@ -70,6 +79,10 @@ class TestAddCommand(AuroraClientCommandTest):
self.TEST_JOBKEY,
self._mock_api,
[2, 3, 4])]
+ assert self._mock_api.query_no_configs.mock_calls == [
+ call(TaskQuery(jobKeys=[self.TEST_JOBKEY.to_thrift()], statuses=ACTIVE_STATES)),
+ call(TaskQuery(jobKeys=[self.TEST_JOBKEY.to_thrift()], statuses=ACTIVE_STATES))
+ ]
def test_no_active_instance(self):
self._fake_context.add_expected_query_result(self.create_empty_task_result())
@@ -79,8 +92,15 @@ class TestAddCommand(AuroraClientCommandTest):
def test_add_instances_raises(self):
self._fake_context.add_expected_query_result(self.create_query_call_result(
self.create_scheduled_task(1, ScheduleStatus.PENDING)))
+ self._fake_context.add_expected_query_result(
+ self.create_query_call_result(), job_key=self.TEST_JOBKEY)
self._mock_api.add_instances.return_value = self.create_error_response()
with pytest.raises(Context.CommandError):
self._command.execute(self._fake_context)
+
+ assert self._mock_api.query_no_configs.mock_calls == [
+ call(TaskQuery(jobKeys=[self.TEST_JOBKEY.to_thrift()], statuses=ACTIVE_STATES)),
+ call(TaskQuery(jobKeys=[self.TEST_JOBKEY.to_thrift()], statuses=ACTIVE_STATES))
+ ]
[3/4] aurora git commit: Enable `Tasks` to specify their own custom
maintenance SLA.
Posted by sa...@apache.org.
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/java/org/apache/aurora/scheduler/state/MaintenanceController.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/state/MaintenanceController.java b/src/main/java/org/apache/aurora/scheduler/state/MaintenanceController.java
deleted file mode 100644
index f58c66a..0000000
--- a/src/main/java/org/apache/aurora/scheduler/state/MaintenanceController.java
+++ /dev/null
@@ -1,294 +0,0 @@
-/**
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.aurora.scheduler.state;
-
-import java.util.Optional;
-import java.util.Set;
-
-import javax.inject.Inject;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Function;
-import com.google.common.base.Predicates;
-import com.google.common.collect.FluentIterable;
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Sets;
-import com.google.common.eventbus.Subscribe;
-
-import org.apache.aurora.gen.HostStatus;
-import org.apache.aurora.gen.MaintenanceMode;
-import org.apache.aurora.gen.ScheduleStatus;
-import org.apache.aurora.scheduler.BatchWorker;
-import org.apache.aurora.scheduler.SchedulerModule.TaskEventBatchWorker;
-import org.apache.aurora.scheduler.base.Query;
-import org.apache.aurora.scheduler.base.Tasks;
-import org.apache.aurora.scheduler.events.PubsubEvent.EventSubscriber;
-import org.apache.aurora.scheduler.events.PubsubEvent.TaskStateChange;
-import org.apache.aurora.scheduler.storage.AttributeStore;
-import org.apache.aurora.scheduler.storage.Storage;
-import org.apache.aurora.scheduler.storage.Storage.MutableStoreProvider;
-import org.apache.aurora.scheduler.storage.entities.IHostAttributes;
-import org.apache.aurora.scheduler.storage.entities.IHostStatus;
-import org.apache.aurora.scheduler.storage.entities.IScheduledTask;
-import org.apache.mesos.v1.Protos;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static java.util.Objects.requireNonNull;
-
-import static org.apache.aurora.gen.MaintenanceMode.DRAINED;
-import static org.apache.aurora.gen.MaintenanceMode.DRAINING;
-
-/**
- * Logic that puts hosts into maintenance mode, and triggers draining of hosts upon request.
- * All state-changing functions return their results. Additionally, all state-changing functions
- * will ignore requests to change state of unknown hosts and subsequently omit these hosts from
- * return values.
- */
-public interface MaintenanceController {
-
- /**
- * Places hosts in maintenance mode.
- * Hosts in maintenance mode are less-preferred for scheduling.
- * No change will be made for hosts that are not recognized, and unrecognized hosts will not be
- * included in the result.
- *
- * @param hosts Hosts to put into maintenance mode.
- * @return The adjusted state of the hosts.
- */
- Set<IHostStatus> startMaintenance(Set<String> hosts);
-
- /**
- * Initiate a drain of all active tasks on {@code hosts}.
- *
- * @param hosts Hosts to drain.
- * @return The adjusted state of the hosts. Hosts without any active tasks will be immediately
- * moved to DRAINED.
- */
- Set<IHostStatus> drain(Set<String> hosts);
-
- /**
- * Drain tasks defined by the inverse offer.
- * This method doesn't set any host attributes.
- *
- * @param inverseOffer the inverse offer to use.
- */
- void drainForInverseOffer(Protos.InverseOffer inverseOffer);
-
- /**
- * Fetches the current maintenance mode of {$code host}.
- *
- * @param host Host to fetch state for.
- * @return Maintenance mode of host, {@link MaintenanceMode#NONE} if the host is not known.
- */
- MaintenanceMode getMode(String host);
-
- /**
- * Fetches the current state of {@code hosts}.
- *
- * @param hosts Hosts to fetch state for.
- * @return The state of the hosts.
- */
- Set<IHostStatus> getStatus(Set<String> hosts);
-
- /**
- * Moves {@code hosts} out of maintenance mode, returning them to mode NONE.
- *
- * @param hosts Hosts to move out of maintenance mode.
- * @return The adjusted state of the hosts.
- */
- Set<IHostStatus> endMaintenance(Set<String> hosts);
-
- class MaintenanceControllerImpl implements MaintenanceController, EventSubscriber {
- private static final Logger LOG = LoggerFactory.getLogger(MaintenanceControllerImpl.class);
- private final Storage storage;
- private final StateManager stateManager;
- private final TaskEventBatchWorker batchWorker;
-
- @Inject
- public MaintenanceControllerImpl(
- Storage storage,
- StateManager stateManager,
- TaskEventBatchWorker batchWorker) {
-
- this.storage = requireNonNull(storage);
- this.stateManager = requireNonNull(stateManager);
- this.batchWorker = requireNonNull(batchWorker);
- }
-
- private Set<String> drainTasksOnHost(String host, MutableStoreProvider store) {
- Query.Builder query = Query.slaveScoped(host).active();
- Set<String> activeTasks = FluentIterable.from(store.getTaskStore().fetchTasks(query))
- .transform(Tasks::id)
- .toSet();
-
- if (activeTasks.isEmpty()) {
- LOG.info("No tasks to drain on host: {}", host);
- // Simple way to avoid the log message if there are no tasks.
- return activeTasks;
- } else {
- LOG.info("Draining tasks: {} on host: {}", activeTasks, host);
- for (String taskId : activeTasks) {
- stateManager.changeState(
- store,
- taskId,
- Optional.empty(),
- ScheduleStatus.DRAINING,
- DRAINING_MESSAGE);
- }
-
- return activeTasks;
- }
-
- }
-
- private Set<IHostStatus> watchDrainingTasks(MutableStoreProvider store, Set<String> hosts) {
- LOG.info("Hosts to drain: " + hosts);
- Set<String> emptyHosts = Sets.newHashSet();
- for (String host : hosts) {
- Set<String> drainedTasks = drainTasksOnHost(host, store);
- // If there are no tasks on the host, immediately transition to DRAINED.
- if (drainedTasks.isEmpty()) {
- emptyHosts.add(host);
- }
- }
-
- return ImmutableSet.<IHostStatus>builder()
- .addAll(setMaintenanceMode(store, emptyHosts, DRAINED))
- .addAll(setMaintenanceMode(store, Sets.difference(hosts, emptyHosts), DRAINING))
- .build();
- }
-
- /**
- * Notifies the MaintenanceController that a task has changed state.
- *
- * @param change Event
- */
- @Subscribe
- public void taskChangedState(final TaskStateChange change) {
- if (Tasks.isTerminated(change.getNewState())) {
- final String host = change.getTask().getAssignedTask().getSlaveHost();
- batchWorker.execute(store -> {
- // If the task _was_ associated with a draining host, and it was the last task on the
- // host.
- Optional<IHostAttributes> attributes =
- store.getAttributeStore().getHostAttributes(host);
- if (attributes.isPresent() && attributes.get().getMode() == DRAINING) {
- Query.Builder builder = Query.slaveScoped(host).active();
- Iterable<IScheduledTask> activeTasks = store.getTaskStore().fetchTasks(builder);
- if (Iterables.isEmpty(activeTasks)) {
- LOG.info("Moving host {} into DRAINED", host);
- setMaintenanceMode(store, ImmutableSet.of(host), DRAINED);
- } else {
- LOG.info("Host {} is DRAINING with active tasks: {}", host, Tasks.ids(activeTasks));
- }
- }
- return BatchWorker.NO_RESULT;
- });
- }
- }
-
- @Override
- public Set<IHostStatus> startMaintenance(Set<String> hosts) {
- return storage.write(
- storeProvider -> setMaintenanceMode(storeProvider, hosts, MaintenanceMode.SCHEDULED));
- }
-
- @VisibleForTesting
- static final Optional<String> DRAINING_MESSAGE =
- Optional.of("Draining machine for maintenance.");
-
- @Override
- public Set<IHostStatus> drain(Set<String> hosts) {
- return storage.write(store -> watchDrainingTasks(store, hosts));
- }
-
- private Optional<String> getHostname(Protos.InverseOffer offer) {
- if (offer.getUrl().getAddress().hasHostname()) {
- return Optional.of(offer.getUrl().getAddress().getHostname());
- } else {
- return Optional.empty();
- }
- }
-
- @Override
- public void drainForInverseOffer(Protos.InverseOffer offer) {
- // TaskStore does not allow for querying by agent id.
- Optional<String> hostname = getHostname(offer);
-
- if (hostname.isPresent()) {
- String host = hostname.get();
- storage.write(storeProvider -> drainTasksOnHost(host, storeProvider));
- } else {
- LOG.error("Unable to drain tasks on agent {} because "
- + "no hostname attached to inverse offer {}.", offer.getAgentId(), offer.getId());
- }
- }
-
- private static final Function<IHostAttributes, String> HOST_NAME =
- IHostAttributes::getHost;
-
- private static final Function<IHostAttributes, IHostStatus> ATTRS_TO_STATUS =
- attributes -> IHostStatus.build(
- new HostStatus().setHost(attributes.getHost()).setMode(attributes.getMode()));
-
- private static final Function<IHostStatus, MaintenanceMode> GET_MODE = IHostStatus::getMode;
-
- @Override
- public MaintenanceMode getMode(final String host) {
- return storage.read(storeProvider -> storeProvider.getAttributeStore().getHostAttributes(host)
- .map(ATTRS_TO_STATUS)
- .map(GET_MODE)
- .orElse(MaintenanceMode.NONE));
- }
-
- @Override
- public Set<IHostStatus> getStatus(final Set<String> hosts) {
- return storage.read(storeProvider -> {
- // Warning - this is filtering _all_ host attributes. If using this to frequently query
- // for a small set of hosts, a getHostAttributes variant should be added.
- return FluentIterable.from(storeProvider.getAttributeStore().getHostAttributes())
- .filter(Predicates.compose(Predicates.in(hosts), HOST_NAME))
- .transform(ATTRS_TO_STATUS)
- .toSet();
- });
- }
-
- @Override
- public Set<IHostStatus> endMaintenance(final Set<String> hosts) {
- return storage.write(
- storeProvider -> setMaintenanceMode(storeProvider, hosts, MaintenanceMode.NONE));
- }
-
- private Set<IHostStatus> setMaintenanceMode(
- MutableStoreProvider storeProvider,
- Set<String> hosts,
- MaintenanceMode mode) {
-
- AttributeStore.Mutable store = storeProvider.getAttributeStore();
- ImmutableSet.Builder<IHostStatus> statuses = ImmutableSet.builder();
- for (String host : hosts) {
- LOG.info("Setting maintenance mode to {} for host {}", mode, host);
- Optional<IHostAttributes> toSave = AttributeStore.Util.mergeMode(store, host, mode);
- if (toSave.isPresent()) {
- store.saveHostAttributes(toSave.get());
- LOG.info("Updated host attributes: " + toSave.get());
- statuses.add(IHostStatus.build(new HostStatus().setHost(host).setMode(mode)));
- }
- }
- return statuses.build();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/java/org/apache/aurora/scheduler/state/StateModule.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/state/StateModule.java b/src/main/java/org/apache/aurora/scheduler/state/StateModule.java
index 0e0f90b..dfc2b5c 100644
--- a/src/main/java/org/apache/aurora/scheduler/state/StateModule.java
+++ b/src/main/java/org/apache/aurora/scheduler/state/StateModule.java
@@ -18,10 +18,8 @@ import javax.inject.Singleton;
import com.beust.jcommander.Parameter;
import com.beust.jcommander.Parameters;
-import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.inject.AbstractModule;
-import com.google.inject.Binder;
import com.google.inject.Module;
import org.apache.aurora.scheduler.app.MoreModules;
@@ -31,7 +29,6 @@ import org.apache.aurora.scheduler.events.PubsubEventModule;
import org.apache.aurora.scheduler.mesos.MesosTaskFactory;
import org.apache.aurora.scheduler.mesos.MesosTaskFactory.MesosTaskFactoryImpl;
import org.apache.aurora.scheduler.scheduling.TaskAssignerImplModule;
-import org.apache.aurora.scheduler.state.MaintenanceController.MaintenanceControllerImpl;
import org.apache.aurora.scheduler.state.UUIDGenerator.UUIDGeneratorImpl;
/**
@@ -69,16 +66,7 @@ public class StateModule extends AbstractModule {
PubsubEventModule.bindSubscriber(binder(), PartitionManager.class);
- bindMaintenanceController(binder());
-
bind(ClusterState.class).to(ClusterStateImpl.class);
bind(ClusterStateImpl.class).in(Singleton.class);
}
-
- @VisibleForTesting
- static void bindMaintenanceController(Binder binder) {
- binder.bind(MaintenanceController.class).to(MaintenanceControllerImpl.class);
- binder.bind(MaintenanceControllerImpl.class).in(Singleton.class);
- PubsubEventModule.bindSubscriber(binder, MaintenanceControllerImpl.class);
- }
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/java/org/apache/aurora/scheduler/thrift/ReadOnlySchedulerImpl.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/thrift/ReadOnlySchedulerImpl.java b/src/main/java/org/apache/aurora/scheduler/thrift/ReadOnlySchedulerImpl.java
index e88cad6..800abfd 100644
--- a/src/main/java/org/apache/aurora/scheduler/thrift/ReadOnlySchedulerImpl.java
+++ b/src/main/java/org/apache/aurora/scheduler/thrift/ReadOnlySchedulerImpl.java
@@ -354,9 +354,13 @@ class ReadOnlySchedulerImpl implements ReadOnlyScheduler.Iface {
public Response getJobUpdateDiff(JobUpdateRequest mutableRequest) {
IJobUpdateRequest request;
try {
- request = IJobUpdateRequest.build(new JobUpdateRequest(mutableRequest).setTaskConfig(
- configurationManager.validateAndPopulate(
- ITaskConfig.build(mutableRequest.getTaskConfig())).newBuilder()));
+ request = IJobUpdateRequest.build(
+ new JobUpdateRequest(mutableRequest)
+ .setTaskConfig(configurationManager
+ .validateAndPopulate(
+ ITaskConfig.build(mutableRequest.getTaskConfig()),
+ mutableRequest.getInstanceCount())
+ .newBuilder()));
} catch (TaskDescriptionException e) {
return error(INVALID_REQUEST, e);
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterface.java
----------------------------------------------------------------------
diff --git a/src/main/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterface.java b/src/main/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterface.java
index 9fc0416..6a28bc2 100644
--- a/src/main/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterface.java
+++ b/src/main/java/org/apache/aurora/scheduler/thrift/SchedulerThriftInterface.java
@@ -63,6 +63,7 @@ import org.apache.aurora.gen.ResourceAggregate;
import org.apache.aurora.gen.Response;
import org.apache.aurora.gen.Result;
import org.apache.aurora.gen.ScheduleStatus;
+import org.apache.aurora.gen.SlaPolicy;
import org.apache.aurora.gen.StartJobUpdateResult;
import org.apache.aurora.gen.StartMaintenanceResult;
import org.apache.aurora.gen.TaskQuery;
@@ -76,11 +77,11 @@ import org.apache.aurora.scheduler.configuration.SanitizedConfiguration;
import org.apache.aurora.scheduler.cron.CronException;
import org.apache.aurora.scheduler.cron.CronJobManager;
import org.apache.aurora.scheduler.cron.SanitizedCronJob;
+import org.apache.aurora.scheduler.maintenance.MaintenanceController;
import org.apache.aurora.scheduler.quota.QuotaCheckResult;
import org.apache.aurora.scheduler.quota.QuotaManager;
import org.apache.aurora.scheduler.quota.QuotaManager.QuotaException;
import org.apache.aurora.scheduler.reconciliation.TaskReconciler;
-import org.apache.aurora.scheduler.state.MaintenanceController;
import org.apache.aurora.scheduler.state.StateChangeResult;
import org.apache.aurora.scheduler.state.StateManager;
import org.apache.aurora.scheduler.state.UUIDGenerator;
@@ -156,6 +157,8 @@ class SchedulerThriftInterface implements AnnotatedAuroraAdmin {
@VisibleForTesting
static final String DRAIN_HOSTS = STAT_PREFIX + "drainHosts";
@VisibleForTesting
+ static final String SLA_DRAIN_HOSTS = STAT_PREFIX + "slaDrainHosts";
+ @VisibleForTesting
static final String MAINTENANCE_STATUS = STAT_PREFIX + "maintenanceStatus";
@VisibleForTesting
static final String END_MAINTENANCE = STAT_PREFIX + "endMaintenance";
@@ -190,6 +193,7 @@ class SchedulerThriftInterface implements AnnotatedAuroraAdmin {
private final AtomicLong restartShardsCounter;
private final AtomicLong startMaintenanceCounter;
private final AtomicLong drainHostsCounter;
+ private final AtomicLong slaDrainHostsCounter;
private final AtomicLong maintenanceStatusCounter;
private final AtomicLong endMaintenanceCounter;
private final AtomicLong addInstancesCounter;
@@ -237,6 +241,7 @@ class SchedulerThriftInterface implements AnnotatedAuroraAdmin {
this.restartShardsCounter = statsProvider.makeCounter(RESTART_SHARDS);
this.startMaintenanceCounter = statsProvider.makeCounter(START_MAINTENANCE);
this.drainHostsCounter = statsProvider.makeCounter(DRAIN_HOSTS);
+ this.slaDrainHostsCounter = statsProvider.makeCounter(SLA_DRAIN_HOSTS);
this.maintenanceStatusCounter = statsProvider.makeCounter(MAINTENANCE_STATUS);
this.endMaintenanceCounter = statsProvider.makeCounter(END_MAINTENANCE);
this.addInstancesCounter = statsProvider.makeCounter(ADD_INSTANCES);
@@ -586,6 +591,14 @@ class SchedulerThriftInterface implements AnnotatedAuroraAdmin {
}
@Override
+ public Response slaDrainHosts(Hosts hosts, SlaPolicy defaultSlaPolicy, long timeoutSecs) {
+ slaDrainHostsCounter.addAndGet(hosts.getHostNamesSize());
+ return ok(Result.drainHostsResult(
+ new DrainHostsResult().setStatuses(IHostStatus.toBuildersSet(
+ maintenance.slaDrain(hosts.getHostNames(), defaultSlaPolicy, timeoutSecs)))));
+ }
+
+ @Override
public Response forceTaskState(String taskId, ScheduleStatus status) {
checkNotBlank(taskId);
requireNonNull(status);
@@ -807,9 +820,12 @@ class SchedulerThriftInterface implements AnnotatedAuroraAdmin {
IJobUpdateRequest request;
try {
- request = IJobUpdateRequest.build(new JobUpdateRequest(mutableRequest).setTaskConfig(
- configurationManager.validateAndPopulate(
- ITaskConfig.build(mutableRequest.getTaskConfig())).newBuilder()));
+ request = IJobUpdateRequest.build(
+ new JobUpdateRequest(mutableRequest)
+ .setTaskConfig(configurationManager.validateAndPopulate(
+ ITaskConfig.build(mutableRequest.getTaskConfig()),
+ mutableRequest.getInstanceCount())
+ .newBuilder()));
} catch (TaskDescriptionException e) {
return error(INVALID_REQUEST, e);
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/python/apache/aurora/admin/admin_util.py
----------------------------------------------------------------------
diff --git a/src/main/python/apache/aurora/admin/admin_util.py b/src/main/python/apache/aurora/admin/admin_util.py
index 8240e80..964876b 100644
--- a/src/main/python/apache/aurora/admin/admin_util.py
+++ b/src/main/python/apache/aurora/admin/admin_util.py
@@ -100,6 +100,29 @@ OVERRIDE_SLA_REASON_OPTION = optparse.Option(
help='Reason for overriding default SLA values. Provide details including the '
'maintenance ticket number.')
+DEFAULT_SLA_PERCENTAGE_OPTION = optparse.Option(
+ '--default_percentage',
+ dest='default_percentage',
+ default=95,
+ help='Percentage of tasks required to be up all the time within the duration. '
+ 'This percentage will be used for SLA calculation if tasks do not '
+ 'have a configured SlaPolicy.')
+
+DEFAULT_SLA_DURATION_OPTION = optparse.Option(
+ '--default_duration',
+ dest='default_duration',
+ default='30m',
+ help='Time interval (now - value) for the percentage of up tasks. Format: XdYhZmWs. '
+ 'This duration will be used for SLA calculation if tasks do not '
+ 'have a configured SlaPolicy.')
+
+FORCE_DRAIN_TIMEOUT_OPTION = optparse.Option(
+ '--force_drain_timeout',
+ dest='timeout',
+ default='7d',
+ help='Time interval (now - value) after which tasks will be forcefully drained. '
+ 'Format: XdYhZmWs.')
+
UNSAFE_SLA_HOSTS_FILE_OPTION = optparse.Option(
'--unsafe_hosts_file',
dest='unsafe_hosts_filename',
@@ -213,6 +236,7 @@ def parse_and_validate_sla_overrides(options, hostnames):
if has_override != all_overrides:
die('All --override_* options are required when attempting to override default SLA values.')
+ print(options.percentage)
percentage = parse_sla_percentage(options.percentage) if options.percentage else None
duration = parse_time(options.duration) if options.duration else None
if options.reason:
@@ -230,6 +254,20 @@ def parse_and_validate_sla_overrides(options, hostnames):
return percentage or SLA_UPTIME_PERCENTAGE_LIMIT, duration or SLA_UPTIME_DURATION_LIMIT
+def parse_and_validate_sla_drain_default(options):
+ """Parses and validates host SLA default 3-tuple (percentage, duration, timeout).
+
+ :param options: command line options
+ :type options: list of app.option
+ :rtype: a tuple of: default percentage (float), default duration (Amount) and timeout (Amount)
+ """
+ percentage = parse_sla_percentage(options.default_percentage)
+ duration = parse_time(options.default_duration).as_(Time.SECONDS)
+ timeout = parse_time(options.timeout).as_(Time.SECONDS)
+
+ return percentage, duration, timeout
+
+
def print_results(results):
"""Prints formatted SLA results.
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/python/apache/aurora/admin/host_maintenance.py
----------------------------------------------------------------------
diff --git a/src/main/python/apache/aurora/admin/host_maintenance.py b/src/main/python/apache/aurora/admin/host_maintenance.py
index 83fc2b6..3ccf322 100644
--- a/src/main/python/apache/aurora/admin/host_maintenance.py
+++ b/src/main/python/apache/aurora/admin/host_maintenance.py
@@ -170,6 +170,7 @@ class HostMaintenance(object):
for hostname in drained_hosts:
callback(hostname)
+ # Deprecated: See AURORA-1986.
def perform_maintenance(self, hostnames, grouping_function=DEFAULT_GROUPING,
percentage=None, duration=None, output_file=None, callback=None):
"""Put hosts into maintenance mode and drain them.
@@ -234,6 +235,25 @@ class HostMaintenance(object):
return set(hostnames) - not_drained_hostnames
+ def perform_sla_maintenance(self, hostnames, percentage, duration, timeout):
+ """Submits hosts for SLA-aware to the scheduler.
+
+
+ :param hostnames: A list of hostnames to operate upon
+ :type hostnames: list of strings
+ :param percentage: SLA percentage to use
+ :type percentage: float
+ :param duration: SLA duration to use
+ :type duration: twitter.common.quantity.Time
+ :param timeout: Force drain timeout
+ :type duration: twitter.common.quantity.Time
+ :rtype: set of host names that were successfully drained
+ """
+ drainable_hosts = Hosts(set(hostnames))
+ check_and_log_response(
+ self._client.sla_drain_hosts(drainable_hosts, percentage, duration, timeout))
+ return drainable_hosts
+
def check_status(self, hostnames):
"""Query the scheduler to determine the maintenance status for a list of hostnames
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/python/apache/aurora/admin/maintenance.py
----------------------------------------------------------------------
diff --git a/src/main/python/apache/aurora/admin/maintenance.py b/src/main/python/apache/aurora/admin/maintenance.py
index 942a237..33a7eea 100644
--- a/src/main/python/apache/aurora/admin/maintenance.py
+++ b/src/main/python/apache/aurora/admin/maintenance.py
@@ -18,13 +18,17 @@ from apache.aurora.client.base import GROUPING_OPTION, get_grouping_or_die, requ
from apache.aurora.common.clusters import CLUSTERS
from .admin_util import (
+ DEFAULT_SLA_DURATION_OPTION,
+ DEFAULT_SLA_PERCENTAGE_OPTION,
FILENAME_OPTION,
+ FORCE_DRAIN_TIMEOUT_OPTION,
HOSTS_OPTION,
OVERRIDE_SLA_DURATION_OPTION,
OVERRIDE_SLA_PERCENTAGE_OPTION,
OVERRIDE_SLA_REASON_OPTION,
POST_DRAIN_SCRIPT_OPTION,
UNSAFE_SLA_HOSTS_FILE_OPTION,
+ parse_and_validate_sla_drain_default,
parse_and_validate_sla_overrides,
parse_hostnames,
parse_script
@@ -132,6 +136,47 @@ def host_drain(cluster):
@app.command
+@app.command_option(FORCE_DRAIN_TIMEOUT_OPTION)
+@app.command_option(FILENAME_OPTION)
+@app.command_option(HOSTS_OPTION)
+@app.command_option(DEFAULT_SLA_PERCENTAGE_OPTION)
+@app.command_option(DEFAULT_SLA_DURATION_OPTION)
+@requires.exactly('cluster')
+def sla_host_drain(cluster):
+ """usage: sla_host_drain {--filename=filename | --hosts=hosts}
+ [--default_percentage=percentage]
+ [--default_duration=duration]
+ [--force_drain_timeout=timeout]
+ cluster
+
+ Asks the scheduler to drain the list of provided hosts in an SLA-aware manner.
+
+ The list of hosts is drained and marked in a drained state. This will kill
+ off any tasks currently running on these hosts, as well as prevent future
+ tasks from scheduling on these hosts while they are drained.
+
+ The hosts are left in maintenance mode upon completion. Use host_activate to
+ return hosts back to service and allow scheduling tasks on them.
+
+ If tasks are unable to be drained after the specified timeout interval they will
+ be forcefully drained even if it breaks SLA.
+ """
+ options = app.get_options()
+ drainable_hosts = parse_hostnames(options.filename, options.hosts)
+
+ percentage, duration, timeout = parse_and_validate_sla_drain_default(options)
+
+ HostMaintenance(
+ cluster=CLUSTERS[cluster],
+ verbosity=options.verbosity,
+ bypass_leader_redirect=options.bypass_leader_redirect).perform_sla_maintenance(
+ drainable_hosts,
+ percentage=percentage,
+ duration=duration,
+ timeout=timeout)
+
+
+@app.command
@app.command_option(FILENAME_OPTION)
@app.command_option(HOSTS_OPTION)
@requires.exactly('cluster')
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/python/apache/aurora/client/api/__init__.py
----------------------------------------------------------------------
diff --git a/src/main/python/apache/aurora/client/api/__init__.py b/src/main/python/apache/aurora/client/api/__init__.py
index f6fd1dd..f1a82cb 100644
--- a/src/main/python/apache/aurora/client/api/__init__.py
+++ b/src/main/python/apache/aurora/client/api/__init__.py
@@ -33,8 +33,10 @@ from gen.apache.aurora.api.ttypes import (
JobUpdateQuery,
JobUpdateRequest,
Metadata,
+ PercentageSlaPolicy,
Resource,
ResourceAggregate,
+ SlaPolicy,
TaskQuery
)
@@ -298,6 +300,16 @@ class AuroraClientAPI(object):
log.info("Draining tasks on: %s" % hosts.hostNames)
return self._scheduler_proxy.drainHosts(hosts)
+ def sla_drain_hosts(self, hosts, percentage=None, duration=None, timeout=None):
+ log.info("Asking scheduler to drain tasks by SLA on: %s" % hosts.hostNames)
+ return self._scheduler_proxy.slaDrainHosts(
+ hosts,
+ SlaPolicy(
+ percentageSlaPolicy=PercentageSlaPolicy(
+ percentage=percentage,
+ durationSecs=duration)),
+ timeout)
+
def maintenance_status(self, hosts):
log.info("Maintenance status for: %s" % hosts.hostNames)
# read-only calls are retriable.
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/python/apache/aurora/client/cli/context.py
----------------------------------------------------------------------
diff --git a/src/main/python/apache/aurora/client/cli/context.py b/src/main/python/apache/aurora/client/cli/context.py
index 06b1941..ad16c2d 100644
--- a/src/main/python/apache/aurora/client/cli/context.py
+++ b/src/main/python/apache/aurora/client/cli/context.py
@@ -244,3 +244,19 @@ class AuroraCommandContext(Context):
raise self.CommandError(EXIT_INVALID_PARAMETER,
"Invalid instance parameter: %s" % (list(unrecognized)))
return active
+
+ def has_count_or_percentage_sla_policy(self, key):
+ """Returns true if any of the tasks has count or percentage sla policy, false otherwise.
+
+ :param key: Job key
+ :type key: AuroraJobKey
+ :return: true if any of the tasks has count or percentage sla policy, false otherwise.
+ """
+ for task in self.get_active_tasks(key):
+ if task.assignedTask.task.slaPolicy:
+ if task.assignedTask.task.slaPolicy.percentageSlaPolicy:
+ return True
+ if task.assignedTask.task.slaPolicy.countSlaPolicy:
+ return True
+
+ return False
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/main/python/apache/aurora/client/cli/jobs.py
----------------------------------------------------------------------
diff --git a/src/main/python/apache/aurora/client/cli/jobs.py b/src/main/python/apache/aurora/client/cli/jobs.py
index 536d04a..f5b6114 100644
--- a/src/main/python/apache/aurora/client/cli/jobs.py
+++ b/src/main/python/apache/aurora/client/cli/jobs.py
@@ -379,6 +379,9 @@ class AddCommand(Verb):
active = context.get_active_instances_or_raise(job, [instance])
start = max(list(active)) + 1
+ if context.has_count_or_percentage_sla_policy(job):
+ context.print_out("WARNING: Adding instances without updating SlaPolicy.")
+
api = context.get_api(job.cluster)
resp = api.add_instances(job, instance, count)
context.log_response_and_raise(resp)
@@ -410,6 +413,10 @@ class KillCommand(AbstractKillCommand):
raise context.CommandError(EXIT_INVALID_PARAMETER,
"The instances list cannot be omitted in a kill command!; "
"use killall to kill all instances")
+
+ if context.has_count_or_percentage_sla_policy(job):
+ context.print_out("WARNING: Killing instances without updating SlaPolicy.")
+
if context.options.strict:
context.get_active_instances_or_raise(job, instances_arg)
api = context.get_api(job.cluster)
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java b/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java
index 63c338e..534a912 100644
--- a/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java
+++ b/src/test/java/org/apache/aurora/scheduler/app/SchedulerIT.java
@@ -39,12 +39,14 @@ import com.google.inject.Guice;
import com.google.inject.Injector;
import com.google.inject.Key;
import com.google.inject.Module;
+import com.google.inject.TypeLiteral;
import org.apache.aurora.GuavaUtils;
import org.apache.aurora.codec.ThriftBinaryCodec.CodingException;
import org.apache.aurora.common.application.Lifecycle;
import org.apache.aurora.common.quantity.Amount;
import org.apache.aurora.common.quantity.Data;
+import org.apache.aurora.common.quantity.Time;
import org.apache.aurora.common.stats.Stats;
import org.apache.aurora.common.zookeeper.Credentials;
import org.apache.aurora.common.zookeeper.testing.BaseZooKeeperTest;
@@ -64,6 +66,7 @@ import org.apache.aurora.scheduler.AppStartup;
import org.apache.aurora.scheduler.TierModule;
import org.apache.aurora.scheduler.base.TaskTestUtil;
import org.apache.aurora.scheduler.config.CliOptions;
+import org.apache.aurora.scheduler.config.types.TimeAmount;
import org.apache.aurora.scheduler.configuration.executor.ExecutorSettings;
import org.apache.aurora.scheduler.discovery.ServiceDiscoveryModule;
import org.apache.aurora.scheduler.discovery.ZooKeeperConfig;
@@ -71,6 +74,7 @@ import org.apache.aurora.scheduler.log.Log;
import org.apache.aurora.scheduler.log.Log.Entry;
import org.apache.aurora.scheduler.log.Log.Position;
import org.apache.aurora.scheduler.log.Log.Stream;
+import org.apache.aurora.scheduler.maintenance.MaintenanceController;
import org.apache.aurora.scheduler.mesos.DriverFactory;
import org.apache.aurora.scheduler.mesos.DriverSettings;
import org.apache.aurora.scheduler.mesos.FrameworkInfoFactory;
@@ -207,6 +211,11 @@ public class SchedulerIT extends BaseZooKeeperTest {
new ServerInfo()
.setClusterName(CLUSTER_NAME)
.setStatsUrlPrefix(STATS_URL_PREFIX)));
+
+ bind(new TypeLiteral<Amount<Long, Time>>() { })
+ .annotatedWith(
+ MaintenanceController.MaintenanceControllerImpl.PollingInterval.class)
+ .toInstance(new TimeAmount(1, Time.MINUTES));
}
};
ZooKeeperConfig zkClientConfig =
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/base/TaskTestUtil.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/base/TaskTestUtil.java b/src/test/java/org/apache/aurora/scheduler/base/TaskTestUtil.java
index 8c1f5ce..abee095 100644
--- a/src/test/java/org/apache/aurora/scheduler/base/TaskTestUtil.java
+++ b/src/test/java/org/apache/aurora/scheduler/base/TaskTestUtil.java
@@ -91,6 +91,8 @@ public final class TaskTestUtil {
true,
true,
true,
+ 2,
+ 1800L,
ConfigurationManager.DEFAULT_ALLOWED_JOB_ENVIRONMENTS);
public static final ExecutorID EXECUTOR_ID = ExecutorID.newBuilder()
.setValue("PLACEHOLDER")
@@ -126,7 +128,7 @@ public final class TaskTestUtil {
.setPriority(1)
.setMaxTaskFailures(-1)
.setProduction(prod)
- .setTier(PROD_TIER_NAME)
+ .setTier(prod ? PROD_TIER_NAME : DEV_TIER_NAME)
.setPartitionPolicy(new PartitionPolicy().setDelaySecs(5).setReschedule(true))
.setSlaPolicy(SlaPolicy.percentageSlaPolicy(
new PercentageSlaPolicy().setPercentage(95.0).setDurationSecs(1800)))
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/config/CommandLineTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/config/CommandLineTest.java b/src/test/java/org/apache/aurora/scheduler/config/CommandLineTest.java
index e66ec11..dcf5889 100644
--- a/src/test/java/org/apache/aurora/scheduler/config/CommandLineTest.java
+++ b/src/test/java/org/apache/aurora/scheduler/config/CommandLineTest.java
@@ -231,9 +231,13 @@ public class CommandLineTest {
expected.mesosLog.coordinatorElectionRetries = 42;
expected.mesosLog.readTimeout = TEST_TIME;
expected.mesosLog.writeTimeout = TEST_TIME;
- expected.sla.slaRefreshInterval = TEST_TIME;
+ expected.sla.minRequiredInstances = 42;
+ expected.sla.maxParallelCoordinators = 42;
+ expected.sla.maxSlaDuration = TEST_TIME;
+ expected.sla.slaCoordinatorTimeout = TEST_TIME;
expected.sla.slaProdMetrics = ImmutableList.of(MetricCategory.JOB_UPTIMES);
expected.sla.slaNonProdMetrics = ImmutableList.of(MetricCategory.JOB_UPTIMES);
+ expected.sla.slaRefreshInterval = TEST_TIME;
expected.webhook.webhookConfigFile = tempFile;
expected.scheduler.maxRegistrationDelay = TEST_TIME;
expected.scheduler.maxLeadingDuration = TEST_TIME;
@@ -251,6 +255,7 @@ public class CommandLineTest {
expected.cron.cronMaxBatchSize = 42;
expected.resourceSettings.enableRevocableCpus = false;
expected.resourceSettings.enableRevocableRam = true;
+ expected.maintenance.hostMaintenancePollingInterval = TEST_TIME;
assertAllNonDefaultParameters(expected);
@@ -385,7 +390,12 @@ public class CommandLineTest {
"-cron_scheduling_max_batch_size=42",
"-enable_revocable_cpus=false",
"-enable_revocable_ram=true",
- "-partition_aware=true"
+ "-partition_aware=true",
+ "-sla_coordinator_timeout=42days",
+ "-host_maintenance_polling_interval=42days",
+ "-max_parallel_coordinated_maintenance=42",
+ "-min_required_instances_for_sla_check=42",
+ "-max_sla_duration_secs=42days"
);
assertEqualOptions(expected, parsed);
}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/configuration/ConfigurationManagerTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/configuration/ConfigurationManagerTest.java b/src/test/java/org/apache/aurora/scheduler/configuration/ConfigurationManagerTest.java
index 749ffea..a99fa7a 100644
--- a/src/test/java/org/apache/aurora/scheduler/configuration/ConfigurationManagerTest.java
+++ b/src/test/java/org/apache/aurora/scheduler/configuration/ConfigurationManagerTest.java
@@ -22,6 +22,8 @@ import com.google.common.collect.ImmutableSet;
import org.apache.aurora.gen.AppcImage;
import org.apache.aurora.gen.Constraint;
import org.apache.aurora.gen.Container;
+import org.apache.aurora.gen.CoordinatorSlaPolicy;
+import org.apache.aurora.gen.CountSlaPolicy;
import org.apache.aurora.gen.CronCollisionPolicy;
import org.apache.aurora.gen.DockerParameter;
import org.apache.aurora.gen.ExecutorConfig;
@@ -33,6 +35,8 @@ import org.apache.aurora.gen.LimitConstraint;
import org.apache.aurora.gen.MesosContainer;
import org.apache.aurora.gen.MesosFetcherURI;
import org.apache.aurora.gen.Mode;
+import org.apache.aurora.gen.PercentageSlaPolicy;
+import org.apache.aurora.gen.SlaPolicy;
import org.apache.aurora.gen.TaskConfig;
import org.apache.aurora.gen.TaskConstraint;
import org.apache.aurora.gen.ValueConstraint;
@@ -74,6 +78,9 @@ public class ConfigurationManagerTest {
private static final ImmutableSet<Container._Fields> ALL_CONTAINER_TYPES =
ImmutableSet.copyOf(Container._Fields.values());
+ private static final int MIN_REQUIRED_INSTANCES = 20;
+ private static final long MAX_SLA_DURATION_SECS = 7200;
+ private static final long DEFAULT_SLA_PERCENTAGE = 95;
private static final JobKey JOB_KEY = new JobKey("owner-role", "devel", "email_stats");
private static final JobConfiguration UNSANITIZED_JOB_CONFIGURATION = new JobConfiguration()
@@ -126,6 +133,8 @@ public class ConfigurationManagerTest {
false,
true,
false,
+ MIN_REQUIRED_INSTANCES,
+ MAX_SLA_DURATION_SECS,
ConfigurationManager.DEFAULT_ALLOWED_JOB_ENVIRONMENTS),
TaskTestUtil.TIER_MANAGER,
TaskTestUtil.THRIFT_BACKFILL,
@@ -139,6 +148,8 @@ public class ConfigurationManagerTest {
true,
true,
true,
+ MIN_REQUIRED_INSTANCES,
+ MAX_SLA_DURATION_SECS,
ConfigurationManager.DEFAULT_ALLOWED_JOB_ENVIRONMENTS),
TaskTestUtil.TIER_MANAGER,
TaskTestUtil.THRIFT_BACKFILL,
@@ -160,7 +171,7 @@ public class ConfigurationManagerTest {
taskConfig.getContainer().getDocker().setImage(null);
expectTaskDescriptionException("A container must specify an image");
- CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(taskConfig));
+ CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(taskConfig), 100);
}
@Test
@@ -169,7 +180,7 @@ public class ConfigurationManagerTest {
taskConfig.getContainer().getDocker().addToParameters(new DockerParameter("foo", "bar"));
expectTaskDescriptionException(ConfigurationManager.NO_DOCKER_PARAMETERS);
- CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(taskConfig));
+ CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(taskConfig), 100);
}
@Test
@@ -179,7 +190,7 @@ public class ConfigurationManagerTest {
builder.unsetExecutorConfig();
expectTaskDescriptionException(ConfigurationManager.EXECUTOR_REQUIRED_WITH_DOCKER);
- CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
+ CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder), 1);
}
@Test
@@ -187,7 +198,7 @@ public class ConfigurationManagerTest {
TaskConfig builder = CONFIG_WITH_CONTAINER.newBuilder();
builder.unsetExecutorConfig();
- DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder), 100);
}
@Test
@@ -196,7 +207,7 @@ public class ConfigurationManagerTest {
.setTier("pr/d"));
expectTaskDescriptionException("Tier contains illegal characters");
- CONFIGURATION_MANAGER.validateAndPopulate(config);
+ CONFIGURATION_MANAGER.validateAndPopulate(config, 100);
}
@Test
@@ -205,7 +216,7 @@ public class ConfigurationManagerTest {
builder.getContainer().getDocker().setParameters(ImmutableList.of());
ITaskConfig result =
- DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder), 100);
// The resulting task config should contain parameters supplied to the ConfigurationManager.
List<IDockerParameter> params = result.getContainer().getDocker().getParameters();
@@ -221,7 +232,7 @@ public class ConfigurationManagerTest {
taskConfig.getContainer().getDocker().addToParameters(userParameter);
ITaskConfig result = DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(
- ITaskConfig.build(taskConfig));
+ ITaskConfig.build(taskConfig), 100);
// The resulting task config should contain parameters supplied from user config.
List<IDockerParameter> params = result.getContainer().getDocker().getParameters();
@@ -236,7 +247,7 @@ public class ConfigurationManagerTest {
.setConstraint(TaskConstraint.value(
new ValueConstraint(false, ImmutableSet.of(JOB_KEY.getRole() + "/f"))))));
- CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
+ CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder), 100);
}
@Test
@@ -246,7 +257,7 @@ public class ConfigurationManagerTest {
.setName(DEDICATED_ATTRIBUTE)
.setConstraint(TaskConstraint.value(new ValueConstraint(false, ImmutableSet.of("*/f"))))));
- CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
+ CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder), 100);
}
@Test
@@ -257,7 +268,7 @@ public class ConfigurationManagerTest {
.setConstraint(TaskConstraint.value(new ValueConstraint(false, ImmutableSet.of("r/f"))))));
expectTaskDescriptionException("Only r may use hosts dedicated for that role.");
- CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
+ CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder), 100);
}
@Test
@@ -267,7 +278,7 @@ public class ConfigurationManagerTest {
builder.addToResources(ramMb(72));
expectTaskDescriptionException("Multiple resource values are not supported for CPU, RAM");
- DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder), 100);
}
@Test
@@ -275,7 +286,7 @@ public class ConfigurationManagerTest {
TaskConfig builder = CONFIG_WITH_CONTAINER.newBuilder();
builder.addToResources(namedPort("thrift"));
- DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder), 100);
}
@Test
@@ -293,10 +304,12 @@ public class ConfigurationManagerTest {
false,
false,
false,
+ MIN_REQUIRED_INSTANCES,
+ MAX_SLA_DURATION_SECS,
ConfigurationManager.DEFAULT_ALLOWED_JOB_ENVIRONMENTS),
TaskTestUtil.TIER_MANAGER,
TaskTestUtil.THRIFT_BACKFILL,
- TestExecutorSettings.THERMOS_EXECUTOR).validateAndPopulate(ITaskConfig.build(builder));
+ TestExecutorSettings.THERMOS_EXECUTOR).validateAndPopulate(ITaskConfig.build(builder), 100);
}
@Test
@@ -312,15 +325,18 @@ public class ConfigurationManagerTest {
new ConfigurationManagerSettings(
ALL_CONTAINER_TYPES,
true,
- ImmutableList.of(new DockerParameter("foo", "bar")),
+ ImmutableList.of(new DockerParameter("foo", "bar")),
false,
false,
false,
false,
+ MIN_REQUIRED_INSTANCES,
+ MAX_SLA_DURATION_SECS,
".+"),
TaskTestUtil.TIER_MANAGER,
TaskTestUtil.THRIFT_BACKFILL,
- TestExecutorSettings.THERMOS_EXECUTOR).validateAndPopulate(ITaskConfig.build(builder));
+ TestExecutorSettings.THERMOS_EXECUTOR)
+ .validateAndPopulate(ITaskConfig.build(builder), 100);
}
@Test
@@ -333,7 +349,7 @@ public class ConfigurationManagerTest {
expectTaskDescriptionException(NO_CONTAINER_VOLUMES);
- CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
+ CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder), 100);
}
@Test
@@ -343,7 +359,7 @@ public class ConfigurationManagerTest {
builder.unsetTaskLinks();
ITaskConfig populated =
- DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder));
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(ITaskConfig.build(builder), 100);
assertEquals(ImmutableSet.of("health", "http"), populated.getTaskLinks().keySet());
}
@@ -361,6 +377,8 @@ public class ConfigurationManagerTest {
true,
true,
true,
+ MIN_REQUIRED_INSTANCES,
+ MAX_SLA_DURATION_SECS,
"b.r"),
TaskTestUtil.TIER_MANAGER,
TaskTestUtil.THRIFT_BACKFILL,
@@ -368,6 +386,166 @@ public class ConfigurationManagerTest {
.validateAndPopulate(IJobConfiguration.build(jobConfiguration));
}
+ @Test
+ public void testCountSlaPolicyUnsupportedTier() throws Exception {
+ ITaskConfig config = ITaskConfig.build(
+ CONFIG_WITH_CONTAINER
+ .newBuilder()
+ .setTier(TaskTestUtil.DEV_TIER_NAME)
+ .setSlaPolicy(SlaPolicy.countSlaPolicy(
+ new CountSlaPolicy()
+ .setCount(MIN_REQUIRED_INSTANCES))));
+
+ expectTaskDescriptionException("Tier 'tier-dev' does not support SlaPolicy.");
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(
+ config,
+ MIN_REQUIRED_INSTANCES);
+ }
+
+ @Test
+ public void testCountSlaPolicyTooFewInstances() throws Exception {
+ ITaskConfig config = ITaskConfig.build(
+ CONFIG_WITH_CONTAINER
+ .newBuilder()
+ .setSlaPolicy(SlaPolicy.countSlaPolicy(
+ new CountSlaPolicy()
+ .setCount(MIN_REQUIRED_INSTANCES))));
+
+ expectTaskDescriptionException(
+ "Job with fewer than 20 instances cannot have Percentage/Count SlaPolicy.");
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(
+ config,
+ MIN_REQUIRED_INSTANCES - 1);
+ }
+
+ @Test
+ public void testCountSlaPolicyCountTooHigh() throws Exception {
+ ITaskConfig config = ITaskConfig.build(
+ CONFIG_WITH_CONTAINER
+ .newBuilder()
+ .setSlaPolicy(SlaPolicy.countSlaPolicy(
+ new CountSlaPolicy()
+ .setCount(MIN_REQUIRED_INSTANCES))));
+
+ expectTaskDescriptionException(
+ "Current CountSlaPolicy: count=20 will not allow any instances to be killed. "
+ + "Must be less than instanceCount=20.");
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(
+ config,
+ MIN_REQUIRED_INSTANCES);
+ }
+
+ @Test
+ public void testCountSlaPolicyDurationSecsTooHigh() throws Exception {
+ ITaskConfig config = ITaskConfig.build(
+ CONFIG_WITH_CONTAINER
+ .newBuilder()
+ .setSlaPolicy(SlaPolicy.countSlaPolicy(
+ new CountSlaPolicy()
+ .setDurationSecs(MAX_SLA_DURATION_SECS + 1))));
+
+ expectTaskDescriptionException("CountSlaPolicy: durationSecs=7201 must be less than "
+ + "cluster-wide maximum of 7200 secs.");
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(
+ config,
+ MIN_REQUIRED_INSTANCES);
+ }
+
+ @Test
+ public void testPercentageSlaPolicyUnsupportedTier() throws Exception {
+ ITaskConfig config = ITaskConfig.build(
+ CONFIG_WITH_CONTAINER
+ .newBuilder()
+ .setTier(TaskTestUtil.DEV_TIER_NAME)
+ .setSlaPolicy(SlaPolicy.percentageSlaPolicy(
+ new PercentageSlaPolicy()
+ .setPercentage(DEFAULT_SLA_PERCENTAGE))));
+
+ expectTaskDescriptionException("Tier 'tier-dev' does not support SlaPolicy.");
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(
+ config,
+ MIN_REQUIRED_INSTANCES);
+ }
+
+ @Test
+ public void testPercentageSlaPolicyTooFewInstances() throws Exception {
+ ITaskConfig config = ITaskConfig.build(
+ CONFIG_WITH_CONTAINER
+ .newBuilder()
+ .setSlaPolicy(SlaPolicy.percentageSlaPolicy(
+ new PercentageSlaPolicy()
+ .setPercentage(DEFAULT_SLA_PERCENTAGE))));
+
+ expectTaskDescriptionException(
+ "Job with fewer than 20 instances cannot have Percentage/Count SlaPolicy.");
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(
+ config,
+ MIN_REQUIRED_INSTANCES - 1);
+ }
+
+ @Test
+ public void testPercentageSlaPolicyPercentageTooHigh() throws Exception {
+ ITaskConfig config = ITaskConfig.build(
+ CONFIG_WITH_CONTAINER
+ .newBuilder()
+ .setSlaPolicy(SlaPolicy.percentageSlaPolicy(
+ new PercentageSlaPolicy()
+ .setPercentage(100))));
+
+ expectTaskDescriptionException(
+ "Current PercentageSlaPolicy: percentage=100.000000 will not allow any instances "
+ + "to be killed. Must be less than 95.000000.");
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(
+ config,
+ MIN_REQUIRED_INSTANCES);
+ }
+
+ @Test
+ public void testPercentageSlaPolicyDurationSecsTooHigh() throws Exception {
+ ITaskConfig config = ITaskConfig.build(
+ CONFIG_WITH_CONTAINER
+ .newBuilder()
+ .setSlaPolicy(SlaPolicy.percentageSlaPolicy(
+ new PercentageSlaPolicy()
+ .setDurationSecs(MAX_SLA_DURATION_SECS + 1))));
+
+ expectTaskDescriptionException("PercentageSlaPolicy: durationSecs=7201 must be less than "
+ + "cluster-wide maximum of 7200 secs.");
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(
+ config,
+ MIN_REQUIRED_INSTANCES);
+ }
+
+ @Test
+ public void testCoordinatorSlaPolicyUnsupportedTier() throws Exception {
+ ITaskConfig config = ITaskConfig.build(
+ CONFIG_WITH_CONTAINER
+ .newBuilder()
+ .setTier(TaskTestUtil.DEV_TIER_NAME)
+ .setSlaPolicy(SlaPolicy.coordinatorSlaPolicy(
+ new CoordinatorSlaPolicy()
+ .setCoordinatorUrl("http://localhost"))));
+
+ expectTaskDescriptionException("Tier 'tier-dev' does not support SlaPolicy.");
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(
+ config,
+ MIN_REQUIRED_INSTANCES);
+ }
+
+ @Test
+ public void testCoordinatorSlaPolicyTooFewInstances() throws Exception {
+ ITaskConfig config = ITaskConfig.build(
+ CONFIG_WITH_CONTAINER
+ .newBuilder()
+ .setSlaPolicy(SlaPolicy.coordinatorSlaPolicy(
+ new CoordinatorSlaPolicy()
+ .setCoordinatorUrl("http://localhost"))));
+
+ DOCKER_CONFIGURATION_MANAGER.validateAndPopulate(
+ config,
+ MIN_REQUIRED_INSTANCES - 1);
+ }
+
private void expectTaskDescriptionException(String message) {
expectedException.expect(TaskDescriptionException.class);
expectedException.expectMessage(message);
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/cron/quartz/CronIT.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/cron/quartz/CronIT.java b/src/test/java/org/apache/aurora/scheduler/cron/quartz/CronIT.java
index 0fabb33..c9ad0cf 100644
--- a/src/test/java/org/apache/aurora/scheduler/cron/quartz/CronIT.java
+++ b/src/test/java/org/apache/aurora/scheduler/cron/quartz/CronIT.java
@@ -66,7 +66,7 @@ public class CronIT extends EasyMockTest {
new JobConfiguration()
.setCronSchedule(CRONTAB_ENTRY.toString())
.setKey(JOB_KEY.newBuilder())
- .setInstanceCount(2)
+ .setInstanceCount(20)
.setOwner(IDENTITY)
.setCronCollisionPolicy(KILL_EXISTING)
.setTaskConfig(makeTaskConfig()));
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/cron/quartz/QuartzTestUtil.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/cron/quartz/QuartzTestUtil.java b/src/test/java/org/apache/aurora/scheduler/cron/quartz/QuartzTestUtil.java
index b7dcf3a..e81c3fa 100644
--- a/src/test/java/org/apache/aurora/scheduler/cron/quartz/QuartzTestUtil.java
+++ b/src/test/java/org/apache/aurora/scheduler/cron/quartz/QuartzTestUtil.java
@@ -37,7 +37,7 @@ final class QuartzTestUtil {
static final IJobConfiguration JOB = IJobConfiguration.build(
new JobConfiguration()
.setCronSchedule("* * * * SUN")
- .setInstanceCount(10)
+ .setInstanceCount(20)
.setOwner(new Identity().setUser("user"))
.setKey(AURORA_JOB_KEY.newBuilder())
.setTaskConfig(TaskTestUtil.makeConfig(AURORA_JOB_KEY)
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/maintenance/MaintenanceControllerImplTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/maintenance/MaintenanceControllerImplTest.java b/src/test/java/org/apache/aurora/scheduler/maintenance/MaintenanceControllerImplTest.java
new file mode 100644
index 0000000..28c62a1
--- /dev/null
+++ b/src/test/java/org/apache/aurora/scheduler/maintenance/MaintenanceControllerImplTest.java
@@ -0,0 +1,422 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.aurora.scheduler.maintenance;
+
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.Executor;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.TypeLiteral;
+
+import org.apache.aurora.common.quantity.Amount;
+import org.apache.aurora.common.quantity.Time;
+import org.apache.aurora.common.stats.StatsProvider;
+import org.apache.aurora.common.testing.easymock.EasyMockTest;
+import org.apache.aurora.gen.CountSlaPolicy;
+import org.apache.aurora.gen.HostAttributes;
+import org.apache.aurora.gen.HostMaintenanceRequest;
+import org.apache.aurora.gen.HostStatus;
+import org.apache.aurora.gen.MaintenanceMode;
+import org.apache.aurora.gen.PercentageSlaPolicy;
+import org.apache.aurora.gen.ScheduledTask;
+import org.apache.aurora.gen.SlaPolicy;
+import org.apache.aurora.scheduler.SchedulerModule.TaskEventBatchWorker;
+import org.apache.aurora.scheduler.async.AsyncModule.AsyncExecutor;
+import org.apache.aurora.scheduler.base.Query;
+import org.apache.aurora.scheduler.base.TaskTestUtil;
+import org.apache.aurora.scheduler.config.types.TimeAmount;
+import org.apache.aurora.scheduler.events.EventSink;
+import org.apache.aurora.scheduler.events.PubsubEvent.TaskStateChange;
+import org.apache.aurora.scheduler.events.PubsubEventModule;
+import org.apache.aurora.scheduler.sla.SlaManager;
+import org.apache.aurora.scheduler.state.PubsubTestUtil;
+import org.apache.aurora.scheduler.state.StateManager;
+import org.apache.aurora.scheduler.storage.Storage;
+import org.apache.aurora.scheduler.storage.entities.IHostAttributes;
+import org.apache.aurora.scheduler.storage.entities.IHostMaintenanceRequest;
+import org.apache.aurora.scheduler.storage.entities.IHostStatus;
+import org.apache.aurora.scheduler.storage.entities.IScheduledTask;
+import org.apache.aurora.scheduler.storage.entities.ISlaPolicy;
+import org.apache.aurora.scheduler.storage.testing.StorageTestUtil;
+import org.apache.aurora.scheduler.testing.FakeStatsProvider;
+import org.apache.mesos.v1.Protos;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.apache.aurora.gen.MaintenanceMode.DRAINED;
+import static org.apache.aurora.gen.MaintenanceMode.DRAINING;
+import static org.apache.aurora.gen.MaintenanceMode.NONE;
+import static org.apache.aurora.gen.MaintenanceMode.SCHEDULED;
+import static org.apache.aurora.gen.ScheduleStatus.KILLED;
+import static org.apache.aurora.gen.ScheduleStatus.RUNNING;
+import static org.apache.aurora.scheduler.testing.BatchWorkerUtil.expectBatchExecute;
+import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.eq;
+import static org.easymock.EasyMock.expect;
+import static org.junit.Assert.assertEquals;
+
+public class MaintenanceControllerImplTest extends EasyMockTest {
+
+ private static final String HOST_A = "a";
+ private static final Set<String> A = ImmutableSet.of(HOST_A);
+ private static final Protos.OfferID OFFER_ID = Protos.OfferID.newBuilder()
+ .setValue("offer-id")
+ .build();
+ private static final Protos.AgentID AGENT_ID = Protos.AgentID.newBuilder()
+ .setValue("agent-id")
+ .build();
+ private static final Protos.FrameworkID FRAMEWORK_ID = Protos.FrameworkID.newBuilder()
+ .setValue("framework-id")
+ .build();
+ private static final Protos.URL AGENT_URL = Protos.URL.newBuilder()
+ .setAddress(Protos.Address.newBuilder()
+ .setHostname(HOST_A)
+ .setPort(5051))
+ .setScheme("http")
+ .build();
+ private static final Protos.Unavailability UNAVAILABILITY = Protos.Unavailability.newBuilder()
+ .setStart(Protos.TimeInfo.newBuilder()
+ .setNanoseconds(Amount.of(1L, Time.MINUTES).as(Time.NANOSECONDS)))
+ .build();
+ private static final SlaPolicy SLA_POLICY = SlaPolicy.percentageSlaPolicy(
+ new PercentageSlaPolicy(95, 1800));
+
+ private static final Protos.InverseOffer INVERSE_OFFER = Protos.InverseOffer.newBuilder()
+ .setId(OFFER_ID)
+ .setAgentId(AGENT_ID)
+ .setUrl(AGENT_URL)
+ .setFrameworkId(FRAMEWORK_ID)
+ .setUnavailability(UNAVAILABILITY)
+ .build();
+
+ private static final SlaPolicy COUNT_SLA_POLICY = SlaPolicy.countSlaPolicy(
+ new CountSlaPolicy()
+ .setCount(2)
+ .setDurationSecs(1800)
+ );
+
+ private StorageTestUtil storageUtil;
+ private StateManager stateManager;
+ private SlaManager slaManager;
+ private MaintenanceController.MaintenanceControllerImpl maintenance;
+ private EventSink eventSink;
+
+ @Before
+ public void setUp() throws Exception {
+ storageUtil = new StorageTestUtil(this);
+ storageUtil.expectOperations();
+ stateManager = createMock(StateManager.class);
+ slaManager = createMock(SlaManager.class);
+ TaskEventBatchWorker batchWorker = createMock(TaskEventBatchWorker.class);
+ expectBatchExecute(batchWorker, storageUtil.storage, control).anyTimes();
+
+ Injector injector = Guice.createInjector(
+ new PubsubEventModule(),
+ new MaintenanceModule(new MaintenanceModule.Options()),
+ new AbstractModule() {
+ @Override
+ protected void configure() {
+ bind(Storage.class).toInstance(storageUtil.storage);
+ bind(StateManager.class).toInstance(stateManager);
+ bind(SlaManager.class).toInstance(slaManager);
+ bind(StatsProvider.class).toInstance(new FakeStatsProvider());
+ bind(Executor.class).annotatedWith(AsyncExecutor.class)
+ .toInstance(MoreExecutors.directExecutor());
+ bind(TaskEventBatchWorker.class).toInstance(batchWorker);
+ bind(new TypeLiteral<Amount<Long, Time>>() { })
+ .annotatedWith(
+ MaintenanceController.MaintenanceControllerImpl.PollingInterval.class)
+ .toInstance(new TimeAmount(1, Time.MINUTES));
+ }
+ });
+ maintenance = injector.getInstance(MaintenanceController.MaintenanceControllerImpl.class);
+ eventSink = PubsubTestUtil.startPubsub(injector);
+ }
+
+ private static IScheduledTask makeTask(String host, String taskId) {
+ ScheduledTask builder = TaskTestUtil.addStateTransition(
+ TaskTestUtil.makeTask(taskId, TaskTestUtil.JOB),
+ RUNNING,
+ 1000).newBuilder();
+ builder.getAssignedTask().setSlaveHost(host);
+ return IScheduledTask.build(builder);
+ }
+
+ @Test
+ public void testMaintenanceCycle() {
+ IScheduledTask task1 = makeTask(HOST_A, "taskA");
+ IScheduledTask task2 = makeTask(HOST_A, "taskB");
+
+ expectMaintenanceModeChange(HOST_A, SCHEDULED);
+ expectFetchTasksByHost(HOST_A, ImmutableSet.of(task1, task2));
+ expectTaskDraining(task1);
+ expectTaskDraining(task2);
+ expectMaintenanceModeChange(HOST_A, DRAINING);
+ IHostAttributes attributes =
+ IHostAttributes.build(new HostAttributes().setHost(HOST_A).setMode(DRAINING));
+ IHostMaintenanceRequest maintenanceRequest =
+ IHostMaintenanceRequest.build(new HostMaintenanceRequest()
+ .setHost(HOST_A)
+ .setCreatedTimestampMs(System.currentTimeMillis())
+ .setDefaultSlaPolicy(SLA_POLICY));
+
+ storageUtil.hostMaintenanceStore.saveHostMaintenanceRequest(
+ anyObject(IHostMaintenanceRequest.class));
+ expect(storageUtil.attributeStore.getHostAttributes(HOST_A))
+ .andReturn(Optional.of(attributes)).times(2);
+ expect(storageUtil.hostMaintenanceStore.getHostMaintenanceRequest(HOST_A))
+ .andReturn(Optional.of(maintenanceRequest)).times(2);
+
+ expect(storageUtil.attributeStore.getHostAttributes()).andReturn(ImmutableSet.of(attributes));
+ expectFetchTasksByHost(HOST_A, ImmutableSet.of(task2));
+ // TaskA is KILLED and therefore no longer active
+ expectFetchTasksByHost(HOST_A, ImmutableSet.of());
+ expectMaintenanceModeChange(HOST_A, DRAINED);
+ expectMaintenanceModeChange(HOST_A, NONE);
+ storageUtil.hostMaintenanceStore.removeHostMaintenanceRequest(HOST_A);
+
+ control.replay();
+
+ assertStatus(HOST_A, SCHEDULED, maintenance.startMaintenance(A));
+ assertStatus(HOST_A, DRAINING, maintenance.drain(A));
+ assertStatus(HOST_A, DRAINING, maintenance.getStatus(A));
+ eventSink.post(
+ TaskStateChange.transition(
+ IScheduledTask.build(task1.newBuilder().setStatus(KILLED)), RUNNING));
+ eventSink.post(
+ TaskStateChange.transition(
+ IScheduledTask.build(task2.newBuilder().setStatus(KILLED)), RUNNING));
+ assertStatus(HOST_A, NONE, maintenance.endMaintenance(A));
+ }
+
+ @Test
+ public void testUnknownHost() {
+ expect(storageUtil.attributeStore.getHostAttributes("b"))
+ .andReturn(Optional.empty());
+
+ control.replay();
+
+ assertEquals(ImmutableSet.of(),
+ maintenance.startMaintenance(ImmutableSet.of("b")));
+ }
+
+ @Test
+ public void testDrainEmptyHost() {
+ storageUtil.hostMaintenanceStore.saveHostMaintenanceRequest(
+ anyObject(IHostMaintenanceRequest.class));
+ expectMaintenanceModeChange(HOST_A, SCHEDULED);
+ expectFetchTasksByHost(HOST_A, ImmutableSet.of());
+ expectMaintenanceModeChange(HOST_A, DRAINED);
+
+ control.replay();
+
+ assertStatus(HOST_A, SCHEDULED, maintenance.startMaintenance(A));
+ assertStatus(HOST_A, DRAINED, maintenance.drain(A));
+ }
+
+ @Test
+ public void testEndEarly() {
+ expectMaintenanceModeChange(HOST_A, SCHEDULED);
+ expectMaintenanceModeChange(HOST_A, NONE);
+ expect(storageUtil.attributeStore.getHostAttributes(HOST_A)).andReturn(Optional.of(
+ IHostAttributes.build(new HostAttributes().setHost(HOST_A).setMode(NONE))));
+ storageUtil.hostMaintenanceStore.removeHostMaintenanceRequest(HOST_A);
+
+ control.replay();
+
+ assertStatus(HOST_A, SCHEDULED, maintenance.startMaintenance(A));
+
+ // End maintenance without DRAINING.
+ assertStatus(HOST_A, NONE, maintenance.endMaintenance(A));
+
+ // Make sure a later transition on the host does not cause any ill effects that could surface
+ // from stale internal state.
+ eventSink.post(TaskStateChange.transition(
+ IScheduledTask.build(makeTask(HOST_A, "taskA").newBuilder().setStatus(KILLED)), RUNNING));
+ }
+
+ @Test
+ public void testSlaDrain() {
+ storageUtil.hostMaintenanceStore.saveHostMaintenanceRequest(
+ anyObject(IHostMaintenanceRequest.class));
+ expectMaintenanceModeChange(HOST_A, DRAINING);
+
+ control.replay();
+
+ assertStatus(
+ HOST_A,
+ DRAINING,
+ maintenance.slaDrain(ImmutableSet.of(HOST_A), COUNT_SLA_POLICY, 1800));
+ }
+
+ @Test
+ public void testSlaDrainUnknownHost() {
+ storageUtil.hostMaintenanceStore.saveHostMaintenanceRequest(
+ anyObject(IHostMaintenanceRequest.class));
+ expect(storageUtil.attributeStore.getHostAttributes("unknown"))
+ .andReturn(Optional.empty());
+
+ control.replay();
+
+ assertEquals(ImmutableSet.of(),
+ maintenance.slaDrain(ImmutableSet.of("unknown"), COUNT_SLA_POLICY, 1800));
+ }
+
+ @Test
+ public void testIteration() {
+ IScheduledTask task1 = makeTask(HOST_A, "taskA");
+ IScheduledTask task2 = makeTask(HOST_A, "taskB");
+
+ IHostAttributes attributes =
+ IHostAttributes.build(new HostAttributes().setHost(HOST_A).setMode(DRAINING));
+ expect(storageUtil.attributeStore.getHostAttributes())
+ .andReturn(ImmutableSet.of(attributes));
+ expectFetchTasksByHost(HOST_A, ImmutableSet.of(task1, task2));
+ IHostMaintenanceRequest maintenanceRequest =
+ IHostMaintenanceRequest.build(new HostMaintenanceRequest()
+ .setHost(HOST_A)
+ .setCreatedTimestampMs(System.currentTimeMillis())
+ .setDefaultSlaPolicy(SLA_POLICY));
+ expect(storageUtil.hostMaintenanceStore.getHostMaintenanceRequest(HOST_A))
+ .andReturn(Optional.of(maintenanceRequest)).times(2);
+ expectTaskDraining(task1);
+ expectTaskDraining(task2);
+
+ control.replay();
+
+ maintenance.runForTest();
+ }
+
+ @Test
+ public void testIterationEmptyHost() {
+ IHostAttributes attributes =
+ IHostAttributes.build(new HostAttributes().setHost(HOST_A).setMode(DRAINING));
+ expect(storageUtil.attributeStore.getHostAttributes())
+ .andReturn(ImmutableSet.of(attributes));
+ expectFetchTasksByHost(HOST_A, ImmutableSet.of());
+ expectMaintenanceModeChange(HOST_A, DRAINED);
+
+ control.replay();
+
+ maintenance.runForTest();
+ }
+
+ @Test
+ public void testIterationMaintenanceTimeout() {
+ IScheduledTask task1 = makeTask(HOST_A, "taskA");
+ IScheduledTask task2 = makeTask(HOST_A, "taskB");
+
+ IHostAttributes attributes =
+ IHostAttributes.build(new HostAttributes().setHost(HOST_A).setMode(DRAINING));
+ expect(storageUtil.attributeStore.getHostAttributes())
+ .andReturn(ImmutableSet.of(attributes));
+ expectFetchTasksByHost(HOST_A, ImmutableSet.of(task1, task2));
+ IHostMaintenanceRequest maintenanceRequest =
+ IHostMaintenanceRequest.build(new HostMaintenanceRequest()
+ .setHost(HOST_A)
+ .setCreatedTimestampMs(0)
+ .setDefaultSlaPolicy(SLA_POLICY));
+ expect(storageUtil.hostMaintenanceStore.getHostMaintenanceRequest(HOST_A))
+ .andReturn(Optional.of(maintenanceRequest)).times(2);
+ expectTaskDraining(task1, true);
+ expectTaskDraining(task2, true);
+
+ control.replay();
+
+ maintenance.runForTest();
+ }
+
+ @Test
+ public void testIterationNoMaintenanceRequest() {
+ IScheduledTask task1 = makeTask(HOST_A, "taskA");
+ IScheduledTask task2 = makeTask(HOST_A, "taskB");
+
+ IHostAttributes attributes =
+ IHostAttributes.build(new HostAttributes().setHost(HOST_A).setMode(DRAINING));
+ expect(storageUtil.attributeStore.getHostAttributes())
+ .andReturn(ImmutableSet.of(attributes));
+ expectFetchTasksByHost(HOST_A, ImmutableSet.of(task1, task2));
+ expect(storageUtil.hostMaintenanceStore.getHostMaintenanceRequest(HOST_A))
+ .andReturn(Optional.empty()).times(2);
+
+ control.replay();
+
+ maintenance.runForTest();
+ }
+
+ @Test
+ public void testGetMode() {
+ expect(storageUtil.attributeStore.getHostAttributes(HOST_A)).andReturn(Optional.of(
+ IHostAttributes.build(new HostAttributes().setHost(HOST_A).setMode(DRAINING))));
+ expect(storageUtil.attributeStore.getHostAttributes("unknown")).andReturn(Optional.empty());
+
+ control.replay();
+
+ assertEquals(DRAINING, maintenance.getMode(HOST_A));
+ assertEquals(NONE, maintenance.getMode("unknown"));
+ }
+
+ @Test
+ public void testInverseOfferDrain() {
+ IScheduledTask task1 = makeTask(HOST_A, "taskA");
+ storageUtil.hostMaintenanceStore.saveHostMaintenanceRequest(
+ anyObject(IHostMaintenanceRequest.class));
+ expectFetchTasksByHost(HOST_A, ImmutableSet.of(task1));
+ expectTaskDraining(task1);
+ IHostMaintenanceRequest maintenanceRequest =
+ IHostMaintenanceRequest.build(new HostMaintenanceRequest()
+ .setHost(HOST_A)
+ .setCreatedTimestampMs(System.currentTimeMillis())
+ .setDefaultSlaPolicy(SLA_POLICY));
+ expect(storageUtil.hostMaintenanceStore.getHostMaintenanceRequest(HOST_A))
+ .andReturn(Optional.of(maintenanceRequest)).times(1);
+
+ control.replay();
+ maintenance.drainForInverseOffer(INVERSE_OFFER);
+ }
+
+ private void expectTaskDraining(IScheduledTask task) {
+ expectTaskDraining(task, false);
+ }
+
+ private void expectTaskDraining(IScheduledTask task, boolean force) {
+ slaManager.checkSlaThenAct(
+ eq(task),
+ eq(ISlaPolicy.build(SLA_POLICY)),
+ anyObject(Storage.MutateWork.class),
+ eq(force));
+ }
+
+ private void expectFetchTasksByHost(String hostName, Set<IScheduledTask> tasks) {
+ expect(storageUtil.taskStore.fetchTasks(Query.slaveScoped(hostName).active())).andReturn(tasks);
+ }
+
+ private void expectMaintenanceModeChange(String hostName, MaintenanceMode mode) {
+ IHostAttributes attributes = IHostAttributes.build(new HostAttributes().setHost(hostName));
+
+ expect(storageUtil.attributeStore.getHostAttributes(hostName))
+ .andReturn(Optional.of(attributes));
+ IHostAttributes updated = IHostAttributes.build(attributes.newBuilder().setMode(mode));
+ expect(storageUtil.attributeStore.saveHostAttributes(updated)).andReturn(true);
+ }
+
+ private void assertStatus(String host, MaintenanceMode mode, Set<IHostStatus> statuses) {
+ assertEquals(ImmutableSet.of(IHostStatus.build(new HostStatus(host, mode))), statuses);
+ }
+}
http://git-wip-us.apache.org/repos/asf/aurora/blob/f2acf53f/src/test/java/org/apache/aurora/scheduler/mesos/MesosCallbackHandlerTest.java
----------------------------------------------------------------------
diff --git a/src/test/java/org/apache/aurora/scheduler/mesos/MesosCallbackHandlerTest.java b/src/test/java/org/apache/aurora/scheduler/mesos/MesosCallbackHandlerTest.java
index c6163bb..8a8c695 100644
--- a/src/test/java/org/apache/aurora/scheduler/mesos/MesosCallbackHandlerTest.java
+++ b/src/test/java/org/apache/aurora/scheduler/mesos/MesosCallbackHandlerTest.java
@@ -36,9 +36,9 @@ import org.apache.aurora.scheduler.base.Conversions;
import org.apache.aurora.scheduler.base.SchedulerException;
import org.apache.aurora.scheduler.events.EventSink;
import org.apache.aurora.scheduler.events.PubsubEvent;
+import org.apache.aurora.scheduler.maintenance.MaintenanceController;
import org.apache.aurora.scheduler.offers.HostOffer;
import org.apache.aurora.scheduler.offers.OfferManager;
-import org.apache.aurora.scheduler.state.MaintenanceController;
import org.apache.aurora.scheduler.storage.Storage;
import org.apache.aurora.scheduler.storage.entities.IHostAttributes;
import org.apache.aurora.scheduler.storage.testing.StorageTestUtil;