codekingpro/portable-devtools

1# Copyright 2021 The Kubernetes Authors.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#     http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14 15 16from . import leaderelection17from .leaderelectionrecord import LeaderElectionRecord18from kubernetes.client.rest import ApiException19from . import electionconfig20import unittest21import threading22import json23import time24import pytest25 26thread_lock = threading.RLock()27 28class LeaderElectionTest(unittest.TestCase):29    def test_simple_leader_election(self):30        election_history = []31        leadership_history = []32 33        def on_create():34            election_history.append("create record")35            leadership_history.append("get leadership")36 37        def on_update():38            election_history.append("update record")39 40        def on_change():41            election_history.append("change record")42 43        mock_lock = MockResourceLock("mock", "mock_namespace", "mock", thread_lock, on_create, on_update, on_change, None)44 45        def on_started_leading():46            leadership_history.append("start leading")47 48        def on_stopped_leading():49            leadership_history.append("stop leading")50 51        # Create config 4.5 4 352        config = electionconfig.Config(lock=mock_lock, lease_duration=2.5,53                                       renew_deadline=2, retry_period=1.5, onstarted_leading=on_started_leading,54                                       onstopped_leading=on_stopped_leading)55 56        # Enter leader election57        leaderelection.LeaderElection(config).run()58 59        self.assert_history(election_history, ["create record", "update record", "update record", "update record"])60        self.assert_history(leadership_history, ["get leadership", "start leading", "stop leading"])61 62    def test_leader_election(self):63        election_history = []64        leadership_history = []65 66        def on_create_A():67            election_history.append("A creates record")68            leadership_history.append("A gets leadership")69 70        def on_update_A():71            election_history.append("A updates record")72 73        def on_change_A():74            election_history.append("A gets leadership")75 76        mock_lock_A = MockResourceLock("mock", "mock_namespace", "MockA", thread_lock, on_create_A, on_update_A, on_change_A, None)77        mock_lock_A.renew_count_max = 378 79        def on_started_leading_A():80            leadership_history.append("A starts leading")81 82        def on_stopped_leading_A():83            leadership_history.append("A stops leading")84 85        config_A = electionconfig.Config(lock=mock_lock_A, lease_duration=2.5,86                                         renew_deadline=2, retry_period=1.5, onstarted_leading=on_started_leading_A,87                                         onstopped_leading=on_stopped_leading_A)88 89        def on_create_B():90            election_history.append("B creates record")91            leadership_history.append("B gets leadership")92 93        def on_update_B():94            election_history.append("B updates record")95 96        def on_change_B():97            leadership_history.append("B gets leadership")98 99        mock_lock_B = MockResourceLock("mock", "mock_namespace", "MockB", thread_lock, on_create_B, on_update_B, on_change_B, None)100        mock_lock_B.renew_count_max = 4101 102        def on_started_leading_B():103            leadership_history.append("B starts leading")104 105        def on_stopped_leading_B():106            leadership_history.append("B stops leading")107 108        config_B = electionconfig.Config(lock=mock_lock_B, lease_duration=2.5,109                                         renew_deadline=2, retry_period=1.5, onstarted_leading=on_started_leading_B,110                                         onstopped_leading=on_stopped_leading_B)111 112        mock_lock_B.leader_record = mock_lock_A.leader_record113 114        threading.daemon = True115        # Enter leader election for A116        threading.Thread(target=leaderelection.LeaderElection(config_A).run()).start()117 118        # Enter leader election for B119        threading.Thread(target=leaderelection.LeaderElection(config_B).run()).start()120 121        time.sleep(5)122 123        self.assert_history(election_history,124                            ["A creates record",125                             "A updates record",126                             "A updates record",127                             "B updates record",128                             "B updates record",129                             "B updates record",130                             "B updates record"])131        self.assert_history(leadership_history,132                            ["A gets leadership",133                             "A starts leading",134                             "A stops leading",135                             "B gets leadership",136                             "B starts leading",137                             "B stops leading"])138 139 140    """Expected behavior: to check if the leader stops leading if it fails to update the lock within the renew_deadline141    and stops leading after finally timing out. The difference between each try comes out to be approximately the sleep142    time.143    Example:144    create record:  0s145    on try update:  1.5s146    on update:  zzz s147    on try update:  3s148    on update: zzz s 149    on try update:  4.5s150    on try update:  6s151    Timeout - Leader Exits"""152    def test_Leader_election_with_renew_deadline(self):153        election_history = []154        leadership_history = []155 156        def on_create():157            election_history.append("create record")158            leadership_history.append("get leadership")159 160        def on_update():161            election_history.append("update record")162 163        def on_change():164            election_history.append("change record")165 166        def on_try_update():167            election_history.append("try update record")168 169        mock_lock = MockResourceLock("mock", "mock_namespace", "mock", thread_lock, on_create, on_update, on_change, on_try_update)170        mock_lock.renew_count_max = 3171 172        def on_started_leading():173            leadership_history.append("start leading")174 175        def on_stopped_leading():176            leadership_history.append("stop leading")177 178        # Create config179        config = electionconfig.Config(lock=mock_lock, lease_duration=2.5,180                                       renew_deadline=2, retry_period=1.5, onstarted_leading=on_started_leading,181                                       onstopped_leading=on_stopped_leading)182 183        # Enter leader election184        leaderelection.LeaderElection(config).run()185 186        self.assert_history(election_history,187                            ["create record",188                             "try update record",189                             "update record",190                             "try update record",191                             "update record",192                             "try update record",193                             "try update record"])194 195        self.assert_history(leadership_history, ["get leadership", "start leading", "stop leading"])196 197    def assert_history(self, history, expected):198        self.assertIsNotNone(expected)199        self.assertIsNotNone(history)200        self.assertEqual(len(expected), len(history))201 202        for idx in range(len(history)):203            self.assertEqual(history[idx], expected[idx],204                              msg="Not equal at index {}, expected {}, got {}".format(idx, expected[idx],205                                                                                      history[idx]))206 207 208class MockResourceLock:209    def __init__(self, name, namespace, identity, shared_lock, on_create=None, on_update=None, on_change=None, on_try_update=None):210        # self.leader_record is shared between two MockResourceLock objects211        self.leader_record = []212        self.renew_count = 0213        self.renew_count_max = 4214        self.name = name215        self.namespace = namespace216        self.identity = str(identity)217        self.lock = shared_lock218 219        self.on_create = on_create220        self.on_update = on_update221        self.on_change = on_change222        self.on_try_update = on_try_update223 224    def get(self, name, namespace):225        self.lock.acquire()226        try:227            if self.leader_record:228                return True, self.leader_record[0]229 230            ApiException.body = json.dumps({'code': 404})231            return False, ApiException232        finally:233            self.lock.release()234 235    def create(self, name, namespace, election_record):236        self.lock.acquire()237        try:238            if len(self.leader_record) == 1:239                return False240            self.leader_record.append(election_record)241            self.on_create()242            self.renew_count += 1243            return True244        finally:245            self.lock.release()246 247    def update(self, name, namespace, updated_record):248        self.lock.acquire()249        try:250            if self.on_try_update:251                self.on_try_update()252            if self.renew_count >= self.renew_count_max:253                return False254 255            old_record = self.leader_record[0]256            self.leader_record[0] = updated_record257 258            self.on_update()259 260            if old_record.holder_identity != updated_record.holder_identity:261                self.on_change()262 263            self.renew_count += 1264            return True265        finally:266            self.lock.release()267 268 269if __name__ == '__main__':270    unittest.main()271