<commit_before># Copyright 2019 Scalyr Inc.
#
# 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.
# ------------------------------------------------------------------------
#
# author: <NAME> <<EMAIL>>
import threading

__author__ = '<EMAIL>'

from scalyr_agent.test_base import ScalyrTestCase
from scalyr_agent.monitor_utils.k8s import DockerMetricFetcher
from scalyr_agent.monitor_utils.k8s import _K8sCache, _K8sProcessor, K8sApiTemporaryError, K8sApiPermanentError, ApiQueryOptions
from scalyr_agent.util import FakeClock
import logging
import time
import mock

from mock import Mock

class Test_K8sCache( ScalyrTestCase ):
    """ Tests the _K8sCache
    """

    NAMESPACE_1 = 'namespace_1'
    POD_1 = 'pod_1'

    class DummyObject(object):
        def __init__( self, access_time ):
            self.access_time = access_time

    def setUp(self):
        self.k8s = FakeK8s()
        self.clock = FakeClock()
        self.processor = FakeProcessor()
        self.cache = _K8sCache( self.processor, 'foo', perform_full_updates=False )

    def tearDown(self):
        self.k8s.stop()

    def test_purge_expired( self ):

        processor = Mock()
        cache = _K8sCache( processor, 'foo', perform_full_updates=False )

        current_time = time.time()
        obj1 = self.DummyObject( current_time - 10 )
        obj2 = self.DummyObject( current_time + 15 )
        obj3 = self.DummyObject( current_time - 20 )

        objects = {
            'default': {
                'obj1': obj1,
                'obj2': obj2,
                'obj3': obj3
            }
        }

        # we should probably look at using actual values returned from k8s here
        # and loading them via 'cache.update'
        cache._objects = objects

        cache.purge_unused(current_time)

        objects = cache._objects.get('default', {})
        self.assertEquals( 1, len( objects ) )
        self.assertTrue( 'obj2' in objects )
        self.assertTrue( objects['obj2'] is obj2 )

    def test_lookup_not_in_cache( self ):

        self.k8s.set_response( self.NAMESPACE_1, self.POD_1, success=True )

        self.assertFalse( self.cache.is_cached( self.NAMESPACE_1, self.POD_1, allow_expired=True ) )

        obj = self.cache.lookup( self.k8s, self.clock.time(), self.NAMESPACE_1, self.POD_1 )

        self.assertTrue( self.cache.is_cached( self.NAMESPACE_1, self.POD_1, allow_expired=True ) )
        self.assertEqual( obj.name, self.POD_1 )
        self.assertEqual( obj.namespace, self.NAMESPACE_1 )

    def test_lookup_already_in_cache( self ):
        query_options = ApiQueryOptions()

        self.k8s.set_response( self.NAMESPACE_1, self.POD_1, success=True )
        obj = self.cache.lookup( self.k8s, self.clock.time(), self.NAMESPACE_1, self.POD_1, query_options=query_options )
        self.assertTrue( self.cache.is_cached( self.NAMESPACE_1, self.POD_1, allow_expired=True ) )

        self.k8s.set_response( self.NAMESPACE_1, self.POD_1, permanent_error=True )
        obj = self.cache.lookup( self.k8s, self.clock.time(), self.NAMESPACE_1, self.POD_1, query_options=query_options )

        self.assertTrue( self.cache.is_cached( self.NAMESPACE_1, self.POD_1, allow_expired=True ) )
        self.assertEqual( obj.name, self.POD_1 )
        self.assertEqual( obj.namespace, self.NAMESPACE_1 )

    def test_raise_exception_on_query_error( self ):
        query_options = ApiQueryOptions()

        self.k8s.set_response( self.NAMESPACE_1, self.POD_1, permanent_error=True )
        self.assertRaises( K8sApiPermanentError, lambda: self.cache.lookup( self.k8s, self.clock.time(), self.NAMESPACE_1, self.POD_1, query_options=query_options ) )

    def test_return_none_on_query_error_with_options( self ):
        query_options = ApiQueryOptions()
        query_options.raise_exception_on_cache_query_error = False

        self.k8s.set_response( self.NAMESPACE_1, self.POD_1, permanent_error=True )
        obj = self.cache.lookup( self.k8s, self.clock.time(), self.NAMESPACE_1, self.POD_1, query_options=query_options )
        self.assertIsNone( obj )

    def test_return_none_on_query_error_without_options( self ):

        self.k8s.set_response( self.NAMESPACE_1, self.POD_1, permanent_error=True )
        obj = self.cache.lookup( self.k8s, self.clock.time(), self.NAMESPACE_1, self.POD_1 )
        self.assertIsNone( obj )


class TestDockerMetricFetcher(ScalyrTestCase):
    """Tests the DockerMetricFetch abstraction.
    """
    def setUp(self):
        self._faker = DockerClientFaker()
        self._fetcher = DockerMetricFetcher(self._faker, 5)

    def test_basic_prefetch(self):
        """Tests the typical prefetch and then get_metrics path.. just for one container.
        """
        self._fetcher.prefetch_metrics('foo')
        self.assertTrue(self._faker.wait_for_requests(1))<commit_msg>added tests for query_api_with_retries<commit_after># Copyright 2019 Scalyr Inc.
#
# 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.
# ------------------------------------------------------------------------
#
# author: <NAME> <<EMAIL>>
import threading

__author__ = '<EMAIL>'

from scalyr_agent.test_base import ScalyrTestCase
from scalyr_agent.monitor_utils.k8s import DockerMetricFetcher
from scalyr_agent.monitor_utils.k8s import _K8sCache, _K8sProcessor, KubernetesApi, K8sApiNotFoundException, K8sApiTemporaryError, K8sApiPermanentError, ApiQueryOptions
from scalyr_agent.monitor_utils.blocking_rate_limiter import BlockingRateLimiter
from scalyr_agent.util import FakeClock
import logging
import time
import mock

from mock import Mock

class Test_K8sCache( ScalyrTestCase ):
    """ Tests the _K8sCache
    """

    NAMESPACE_1 = 'namespace_1'
    POD_1 = 'pod_1'

    class DummyObject(object):
        def __init__( self, access_time ):
            self.access_time = access_time

    def setUp(self):
        self.k8s = FakeK8s()
        self.clock = FakeClock()
        self.processor = FakeProcessor()
        self.cache = _K8sCache( self.processor, 'foo', perform_full_updates=False )

    def tearDown(self):
        self.k8s.stop()

    def test_purge_expired( self ):

        processor = Mock()
        cache = _K8sCache( processor, 'foo', perform_full_updates=False )

        current_time = time.time()
        obj1 = self.DummyObject( current_time - 10 )
        obj2 = self.DummyObject( current_time + 15 )
        obj3 = self.DummyObject( current_time - 20 )

        objects = {
            'default': {
                'obj1': obj1,
                'obj2': obj2,
                'obj3': obj3
            }
        }

        # we should probably look at using actual values returned from k8s here
        # and loading them via 'cache.update'
        cache._objects = objects

        cache.purge_unused(current_time)

        objects = cache._objects.get('default', {})
        self.assertEquals( 1, len( objects ) )
        self.assertTrue( 'obj2' in objects )
        self.assertTrue( objects['obj2'] is obj2 )

    def test_lookup_not_in_cache( self ):

        self.k8s.set_response( self.NAMESPACE_1, self.POD_1, success=True )

        self.assertFalse( self.cache.is_cached( self.NAMESPACE_1, self.POD_1, allow_expired=True ) )

        obj = self.cache.lookup( self.k8s, self.clock.time(), self.NAMESPACE_1, self.POD_1 )

        self.assertTrue( self.cache.is_cached( self.NAMESPACE_1, self.POD_1, allow_expired=True ) )
        self.assertEqual( obj.name, self.POD_1 )
        self.assertEqual( obj.namespace, self.NAMESPACE_1 )

    def test_lookup_already_in_cache( self ):
        query_options = ApiQueryOptions()

        self.k8s.set_response( self.NAMESPACE_1, self.POD_1, success=True )
        obj = self.cache.lookup( self.k8s, self.clock.time(), self.NAMESPACE_1, self.POD_1, query_options=query_options )
        self.assertTrue( self.cache.is_cached( self.NAMESPACE_1, self.POD_1, allow_expired=True ) )

        self.k8s.set_response( self.NAMESPACE_1, self.POD_1, permanent_error=True )
        obj = self.cache.lookup( self.k8s, self.clock.time(), self.NAMESPACE_1, self.POD_1, query_options=query_options )

        self.assertTrue( self.cache.is_cached( self.NAMESPACE_1, self.POD_1, allow_expired=True ) )
        self.assertEqual( obj.name, self.POD_1 )
        self.assertEqual( obj.namespace, self.NAMESPACE_1 )

    def test_raise_exception_on_query_error( self ):
        query_options = ApiQueryOptions()

        self.k8s.set_response( self.NAMESPACE_1, self.POD_1, permanent_error=True )
        self.assertRaises( K8sApiPermanentError, lambda: self.cache.lookup( self.k8s, self.clock.time(), self.NAMESPACE_1, self.POD_1, query_options=query_options ) )

    def test_return_none_on_query_error_with_options( self ):
        query_options = ApiQueryOptions()
        query_options.raise_exception_on_cache_query_error = False

        self.k8s.set_response( self.NAMESPACE_1, self.POD_1, permanent_error=True )
        obj = self.cache.lookup( self.k8s, self.clock.time(), self.NAMESPACE_1, self.POD_1, query_options=query_options )
        self.assertIsNone( obj )

    def test_return_none_on_query_error_without_options( self ):

        self.k8s.set_response( self.NAMESPACE_1, self.POD_1, permanent_error=True )
        obj = self.cache.lookup( self.k8s, self.clock.time(), self.NAMESPACE_1, self.POD_1 )
        self.assertIsNone( obj )

class TestKubernetesApi( ScalyrTestCase ):
    """
    Tests the Kubernetes API
    """

    def test_query_api_with_retries_success_not_rate_limited( self ):
        with mock.patch.object( KubernetesApi, "query_api" ) as mock_query:
            mock_query.return_value = { "success": "success" }

            k8s = KubernetesApi()
            rate_limiter = BlockingRateLimiter(
                num_agents=1, initial_cluster_rate=100, max_cluster_rate=1000, min_cluster_rate=1,
                consecutive_success_threshold=1,
                strategy='multiply',
            )
            options = ApiQueryOptions( rate_limiter=rate_limiter )
            result = k8s.query_api_with_retries( "/foo/bar", options )
            self.assertEqual( result, { "success": "success" } )
            self.assertEqual( rate_limiter.current_cluster_rate, 200.0 )

    def test_query_api_with_retries_not_found_not_rate_limited( self ):
        with mock.patch.object( KubernetesApi, "query_api" ) as mock_query:
            mock_query.side_effect = K8sApiNotFoundException( "/foo/bar" )

            k8s = KubernetesApi()
            rate_limiter = BlockingRateLimiter(
                num_agents=1, initial_cluster_rate=100, max_cluster_rate=1000, min_cluster_rate=1,
                consecutive_success_threshold=1,
                strategy='multiply',
            )
            options = ApiQueryOptions( rate_limiter=rate_limiter )
            self.assertRaises( K8sApiNotFoundException, lambda: k8s.query_api_with_retries( "/foo/bar", options ) )
            self.assertEqual( rate_limiter.current_cluster_rate, 200.0 )

    def test_query_api_with_retries_temp_error_rate_limited( self ):
        with mock.patch.object( KubernetesApi, "query_api" ) as mock_query:
            mock_query.side_effect = K8sApiTemporaryError( "Temporary Error" )

            k8s = KubernetesApi()
            rate_limiter = BlockingRateLimiter(
                num_agents=1, initial_cluster_rate=100, max_cluster_rate=1000, min_cluster_rate=1,
                consecutive_success_threshold=1,
                strategy='multiply',
            )
            options = ApiQueryOptions( rate_limiter=rate_limiter, max_retries=0 )
            self.assertRaises( K8sApiTemporaryError, lambda: k8s.query_api_with_retries( "/foo/bar", options ) )
            self.assertEqual( rate_limiter.current_cluster_rate, 50.0 )

    def test_query_api_with_retries_other_error_rate_limited( self ):
        with mock.patch.object( KubernetesApi, "query_api" ) as mock_query:
            mock_query.side_effect = K8sApiPermanentError( "Permanent Error" )

            k8s = KubernetesApi()
            rate_limiter = BlockingRateLimiter(
                num_agents=1, initial_cluster_rate=100, max_cluster_rate=1000, min_cluster_rate=1,
                consecutive_success_threshold=1,
                strategy='multiply',
            )
            options = ApiQueryOptions( rate_limiter=rate_limiter, max_retries=0 )
            self.assertRaises( K8sApiPermanentError, lambda: k8s.query_api_with_retries( "/foo/bar", options ) )
            self.assertEqual( rate_limiter.current_cluster_rate, 50.0 )

            mock_query.side_effect = Exception( "Some other exception" )
            self.assertRaises( Exception, lambda: k8s.query_api_with_retries( "/foo/bar", options ) )
            self.assertEqual( rate_limiter.current_cluster_rate, 25.0 )

class TestDockerMetricFetcher(ScalyrTestCase):
    """Tests the DockerMetricFetch abstraction.
    """
    def setUp(self):
        self._faker = DockerClientFaker()
        self._fetcher = DockerMetricFetcher(self._faker, 5)

    def test_basic_prefetch(self):
        """Tests the typical prefetch and then get_metrics path.. just for one container.
        """
        self._fetcher.prefetch_metrics('foo')
        self.assertTrue(self._faker.wait_for_requests(1))
{'fraction_non_alphanumeric': 0.1040332230590554, 'fraction_numerical': 0.01879355247413247, 'mean_word_length': 2.9776035834266517, 'pattern_counts': {'":': 2, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2548882', 'n_tokens_mistral': 4540, 'n_tokens_neox': 4456, 'n_words': 919}